From b6f54633e5b4cafa851fdec80e8e982d46109be3 Mon Sep 17 00:00:00 2001
From: Helmut Grohne <helmut@subdivi.de>
Date: Tue, 18 May 2021 15:43:14 +0200
Subject: mmdebstrap: reimplement build hook recursively

Instead of writing a complex shell script in a limited sub-language of
shell, call out to ourselves and implement the build functionality in
Python.
---
 mdbp/common.py     |  40 +++++----
 mdbp/mmdebstrap.py | 257 +++++++++++++++++++++++++++++------------------------
 2 files changed, 166 insertions(+), 131 deletions(-)

diff --git a/mdbp/common.py b/mdbp/common.py
index 830aad7..eccefc1 100644
--- a/mdbp/common.py
+++ b/mdbp/common.py
@@ -119,6 +119,27 @@ def download(uri: str, checksums: typing.Dict[str, str],
             dest.unlink()
             raise
 
+def download_dsc(buildinput: JsonObject,
+                 destdir: pathlib.Path) -> pathlib.Path:
+    """Download the .input.dscuri including referenced components to the given
+    destination directory and return the path to the contained .dsc file.
+    """
+    dscuri = buildinput["dscuri"]
+    dscpath = destdir / dscuri.split("/")[-1]
+    # mypy doesn't grok this:
+    assert isinstance(dscpath, pathlib.Path)
+    download(dscuri, buildinput.get("checksums", {}), dscpath)
+    files: typing.Dict[str, typing.Dict[str, str]] = {}
+    with dscpath.open("r") as dscf:
+        for key, value in debian.deb822.Dsc(dscf).items():
+            if key.lower().startswith("checksums-"):
+                for entry in value:
+                    algo = key[10:].lower()
+                    files.setdefault(entry["name"], dict())[algo] = entry[algo]
+    for name, checksums in files.items():
+        download(urllib.parse.urljoin(dscuri, name), checksums, destdir / name)
+    return dscpath
+
 @contextlib.contextmanager
 def get_dsc(build: JsonObject) -> typing.Iterator[pathlib.Path]:
     """A context manager that provides a path pointing at the .dsc file for the
@@ -129,23 +150,8 @@ def get_dsc(build: JsonObject) -> typing.Iterator[pathlib.Path]:
     try:
         dscpath = build["input"]["dscpath"]
     except KeyError:
-        dscuri = build["input"]["dscuri"]
-        with tempfile.TemporaryDirectory() as tdirname:
-            tdir = pathlib.Path(tdirname)
-            dscpath = tdir / dscuri.split("/")[-1]
-            download(dscuri, build["input"].get("checksums", {}), dscpath)
-            files: typing.Dict[str, typing.Dict[str, str]] = {}
-            with dscpath.open("r") as dscf:
-                for key, value in debian.deb822.Dsc(dscf).items():
-                    if key.lower().startswith("checksums-"):
-                        for entry in value:
-                            algo = key[10:].lower()
-                            files.setdefault(entry["name"], dict())[algo] = \
-                                    entry[algo]
-            for name, checksums in files.items():
-                download(urllib.parse.urljoin(dscuri, name), checksums,
-                         tdir / name)
-            yield dscpath
+        with tempfile.TemporaryDirectory() as tdir:
+            yield download_dsc(build["input"], pathlib.Path(tdir))
     else:
         yield pathlib.Path(dscpath)
 
diff --git a/mdbp/mmdebstrap.py b/mdbp/mmdebstrap.py
index 5a70540..62d3195 100644
--- a/mdbp/mmdebstrap.py
+++ b/mdbp/mmdebstrap.py
@@ -3,60 +3,142 @@
 """mdbp backend using mmdebstrap"""
 
 import argparse
+import ctypes
+import ctypes.util
+import functools
+import os
 import pathlib
+import pwd
 import shlex
+import shutil
 import subprocess
 import sys
-import tempfile
 import typing
 
-from .common import buildjson, compute_env, get_dsc, get_dsc_files, \
-        profile_option
+from .common import buildjson, compute_env, download_dsc, get_dsc_files, \
+        json_load, profile_option
 
-# pylint: disable=W0102 # as we do not modify env
-def priv_drop(cmd: typing.List[str], *, chroot: bool = False,
+libc = ctypes.CDLL(ctypes.util.find_library("c"))
+def unshare_network() -> None:
+    """The unshare(2) the network namespace. Errors are raised as OSError."""
+    if libc.unshare(0x40000000) < 0:
+        raise OSError("unshare() failed", ctypes.get_errno())
+
+def priv_drop(cmd: typing.List[str], *,
+              chroot: typing.Optional[pathlib.Path] = None,
               chdir: typing.Union[None, str, pathlib.PurePath] = None,
               setuid: typing.Optional[str] = None,
               privnet: bool = False,
-              env: typing.Dict[str, str] = {}) -> str:
-    """Returns shlext.join(cmd) with certain privilege dropping operations
-    applied:
-     * When chroot is True, chroot to "$1".
+              env: typing.Optional[typing.Dict[str, str]] = None) -> None:
+    """Invoke the given command as a subprocess with the given confinements.
+     * A chroot target can be specified.
      * When chdir is set, chdir to it after chrooting. Should be absolute.
      * When setuid is set, change the user id and group id. Names allowed.
      * When privnet is True, create a new network namespace with an active
        loopback interface.
      * Environment variables from env will be set.
+    A non-zero exit code is raised as a CalledProcessError.
     """
-    if chdir or env:
-        envcmd = ["env"]
-        if chdir:
-            envcmd.extend(["--chdir", str(chdir)])
-        envcmd.extend(map("%s=%s".__mod__, env.items()))
-        cmd = envcmd + cmd
-    cmdstring = ""
-    if setuid or chroot:
-        cmdstring = "chroot "
-        if setuid:
-            cmdstring += "--userspec %s " % \
-                    shlex.quote("%s:%s" % (setuid, setuid))
-        cmdstring += '"$1" ' if chroot else "/ "
-    cmdstring += shlex.join(cmd)
-    if privnet:
-        cmd = ["unshare", "--net",
-               "sh", "-c", "ip link set dev lo up && " + cmdstring]
-        cmdstring = shlex.join(cmd)
+    def preexec_fn() -> None:
+        if privnet:
+            unshare_network()
+            subprocess.check_call(["ip", "link", "set", "dev", "lo", "up"])
         if chroot:
-            cmdstring += ' exec "$1"'
-    return cmdstring
+            os.chroot(chroot)
+        if chroot or chdir:
+            os.chdir(chdir or "/")
+        if setuid:
+            pwentry = pwd.getpwnam(setuid)
+            os.setgid(pwentry.pw_gid)
+            os.setuid(pwentry.pw_uid)
+    subprocess.check_call(cmd, preexec_fn=preexec_fn, env=env)
+
+def native_architecture() -> str:
+    """Return the native architecture as returned by dpkg."""
+    return subprocess.check_output(["dpkg", "--print-architecture"],
+                                   encoding="ascii").strip()
+
+def hook_main(buildjsonfilename: str, chrootname: str) -> None:
+    """The entry point for the --hook-helper invocation run from mmdebstrap."""
+    build = json_load(pathlib.Path(buildjsonfilename).open("r"))
+    chroot = pathlib.Path(chrootname)
+    buildpath = pathlib.PurePath(build.get("buildpath", "/build/build"))
+    fullbuildpath = chroot / buildpath.relative_to("/")
+    if "dscpath" in build["input"]:
+        dscpath = fullbuildpath.parent / \
+                pathlib.PurePath(build["input"]["dscpath"]).name
+    else:
+        dscpath = download_dsc(build["input"], fullbuildpath.parent)
+    priv_drop(["chown", "-R", "build:build", "."],
+               chroot=chroot, chdir=buildpath.parent)
+    priv_drop(["dpkg-source", "--no-check", "--extract", dscpath.name,
+               buildpath.name],
+              setuid="build", chroot=chroot, chdir=buildpath.parent)
+    for path in [*get_dsc_files(dscpath), dscpath]:
+        path.unlink()
+    hostarch = build.get("hostarch") or build.get("buildarch") or \
+            native_architecture()
+    apt_get = ["apt-get", "--yes", "-oAPT::Keep-Downloaded-Packages=false"]
+    cmd = [*apt_get, "build-dep", "--host-architecture", hostarch,
+           *dict(any=["--arch-only"],
+                 all=["--indep-only"]).get(build.get("type"), ()),
+           *profile_option(build, "--build-profiles"),
+           "./"]
+    try:
+        priv_drop(cmd, chroot=chroot, chdir=buildpath)
+    except subprocess.CalledProcessError:
+        if build.get("bd-uninstallable-explainer") != "apt" or \
+           not build["output"].get("log", True):
+            raise
+        cmd[-1:-1] = ['-oDebug::pkgProblemResolver=true',
+                      '-oDebug::pkgDepCache::Marker=1',
+                      '-oDebug::pkgDepCache::AutoInstall=1',
+                      '-oDebug::BuildDeps=1']
+        priv_drop(cmd, chroot=chroot, chdir=buildpath)
+    priv_drop(["dpkg-buildpackage", "-uc", "--host-arch=" + hostarch,
+               "--build=" + build.get("type", "binary"),
+               *profile_option(build, "--build-profiles=")],
+              chroot=chroot, setuid="build",
+              privnet=not build.get("network") in ("enable", "try-enable"),
+              chdir=buildpath, env=compute_env(build))
+    shutil.rmtree(fullbuildpath)
+    if build.get("lintian", {}).get("run"):
+        priv_drop([*apt_get, "install", "lintian"], chroot=chroot)
+        priv_drop(["lintian", *build["lintian"].get("options", ()),
+                   "%s_%s.changes" % (dscpath.stem, hostarch)],
+                  chroot=chroot, setuid="build", chdir=buildpath.parent)
+
+class RawStoreAction(argparse.Action):
+    """An action that stores the raw value in addition to the type-parsed
+    value. An additional "raw_"-prefixed attribute is added to the namespace
+    carrying the value prior to passing it through the type function."""
+
+    # pylint: disable=W0622
+    def __init__(self, *args: typing.Any,
+                 type: typing.Callable[[str], typing.Any],
+                 **kwargs: typing.Any) -> None:
+        def raw_type(value: typing.Any) -> typing.Any:
+            return (value, type(value))
+        kwargs["type"] = functools.wraps(type)(raw_type)
+        super().__init__(*args, **kwargs)
+
+    def __call__(self, parser: argparse.ArgumentParser,
+                 namespace: argparse.Namespace, rawvalue: typing.Any,
+                 option_string: typing.Optional[str]=None) -> None:
+        setattr(namespace, "raw_" + self.dest, rawvalue[0])
+        setattr(namespace, self.dest, rawvalue[1])
 
 def main() -> None:
     """Entry point for mdbp-mmdebstrap backend"""
+    if len(sys.argv) == 4 and sys.argv[1] == "--hook-helper":
+        hook_main(sys.argv[2], sys.argv[3])
+        return
+
     parser = argparse.ArgumentParser()
     parser.add_argument("--mirror", type=str, action="store",
                         default="http://deb.debian.org/debian",
                         help="mirror url to fetch packages from")
-    parser.add_argument("buildjson", type=buildjson)
+    parser.add_argument("buildjson", type=buildjson, action=RawStoreAction)
     args = parser.parse_args()
     build = args.buildjson
 
@@ -64,9 +146,7 @@ def main() -> None:
         raise ValueError("bd-uinstallable-explainer %r not supported" %
                          build.get("bd-uinstallable-explainer"))
 
-    buildarch = build.get("buildarch") or \
-        subprocess.check_output(["dpkg", "--print-architecture"],
-                                encoding="ascii").strip()
+    buildarch = build.get("buildarch") or native_architecture()
     hostarch = build.get("hostarch") or buildarch
 
     if buildarch == hostarch:
@@ -79,87 +159,36 @@ def main() -> None:
     buildpath = pathlib.PurePath(build.get("buildpath", "/build/build"))
     enablelog = build["output"].get("log", True)
 
-    with get_dsc(build) as dscpath, \
-         tempfile.NamedTemporaryFile("w+") as script:
-        dscfiles = [dscpath] + get_dsc_files(dscpath)
-
-        script.write("#!/bin/sh\nset -u\nset -e\n%s\n%s\n%s\n" % (
-            priv_drop(["chown", "-R", "build:build", "."],
-                      chroot=True, chdir=buildpath.parent),
-            priv_drop(["dpkg-source", "--no-check", "--extract", dscpath.name,
-                       buildpath.name],
-                      setuid="build", chroot=True, chdir=buildpath.parent),
-            priv_drop(["rm", *(f.name for f in dscfiles)],
-                      chroot=True, chdir=buildpath.parent)))
-        apt_get = ["apt-get", "--yes", "-oAPT::Keep-Downloaded-Packages=false"]
-        cmd = [*apt_get, "build-dep", "--host-architecture", hostarch,
-               *dict(any=["--arch-only"],
-                     all=["--indep-only"]).get(build.get("type"), ()),
-               *profile_option(build, "--build-profiles"),
-               "./"]
-        if build.get("bd-uninstallable-explainer") == "apt" and enablelog:
-            script.write("if ! %s\nthen\n" % priv_drop(cmd, chroot=True,
-                                                       chdir=buildpath))
-            cmd[-1:-1] = ['-oDebug::pkgProblemResolver=true',
-                          '-oDebug::pkgDepCache::Marker=1',
-                          '-oDebug::pkgDepCache::AutoInstall=1',
-                          '-oDebug::BuildDeps=1']
-            script.write("%s\nfi\n" % priv_drop(cmd, chroot=True,
-                                                chdir=buildpath))
-        else:
-            script.write("%s\n" % priv_drop(cmd, chroot=True, chdir=buildpath))
-        cmd = ["dpkg-buildpackage", "-uc", "--host-arch=" + hostarch,
-               "--build=" + build.get("type", "binary")]
-        cmd.extend(profile_option(build, "--build-profiles="))
-        script.write("%s\n" % priv_drop(
-            cmd, chroot=True, setuid="build",
-            privnet=not build.get("network") in ("enable", "try-enable"),
-            chdir=buildpath, env=compute_env(build)))
-        script.write("%s\n" % priv_drop(["rm", "-R", str(buildpath)],
-                                        chroot=True))
-        if build.get("lintian", {}).get("run"):
-            cmd = ["lintian", *build["lintian"].get("options", ()),
-                   "%s_%s.changes" % (dscpath.stem, hostarch)]
-            script.write("%s\n%s\n" %
-                         (priv_drop([*apt_get, "install", "lintian"],
-                                    chroot=True),
-                          priv_drop(cmd, chroot=True, setuid="build",
-                                    chdir=buildpath.parent)))
-        # Only close the file object, script.close would delete it.
-        # Unfortunatly, mypy doesn't grok the indirection and thinks that
-        # script already is the file.
-        script.file.close() # type: ignore[attr-defined]
-        # World readable as the hook may be run by a different uid.
-        pathlib.Path(script.name).chmod(0o755)
-        cmd = [
-            "mmdebstrap",
-            "--verbose" if enablelog else "--quiet",
-            "--mode=unshare",
-            "--variant=apt",
-            "--architectures=" +
-                ",".join(dict.fromkeys((buildarch, hostarch))),
-            "--include=" + ",".join(buildessential),
-            '--customize-hook=chroot "$1" useradd --user-group --create-home '
-                '--home-dir %s build --skel /nonexistent' %
-                shlex.quote(str(buildpath.parent)),
-            "--customize-hook=copy-in %s %s" %
-                (shlex.join(map(str, dscfiles)),
-                 shlex.quote(str(buildpath.parent))),
-            "--customize-hook=" + script.name,
-            "--customize-hook=sync-out " +
-                shlex.join([str(buildpath.parent),
-                            build["output"]["directory"]]),
-            build["distribution"],
-            "/dev/null",
-            args.mirror,
-        ]
-        cmd.extend(build.get("extrarepositories", ()))
-        proc = subprocess.Popen(cmd,
-                                stdout=None if enablelog
-                                       else subprocess.DEVNULL,
-                                stderr=subprocess.STDOUT if enablelog
-                                       else subprocess.DEVNULL)
-        sys.exit(proc.wait())
+    cmd = [
+        "mmdebstrap",
+        "--verbose" if enablelog else "--quiet",
+        "--mode=unshare",
+        "--variant=apt",
+        "--architectures=" +
+            ",".join(dict.fromkeys((buildarch, hostarch))),
+        "--include=" + ",".join(buildessential),
+        '--customize-hook=chroot "$1" useradd --user-group --create-home '
+            '--home-dir %s build --skel /nonexistent' %
+            shlex.quote(str(buildpath.parent)),
+        *(["--customize-hook=copy-in " + shlex.join([
+            build["input"]["dscpath"],
+            *map(str, get_dsc_files(pathlib.Path(build["input"]["dscpath"]))),
+            str(buildpath.parent)])]
+          if "dscpath" in build["input"] else ()),
+        '--customize-hook=mdbp-mmdebstrap --hook-helper %s "$1"' %
+            shlex.quote(args.raw_buildjson),
+        "--customize-hook=sync-out " +
+            shlex.join([str(buildpath.parent), build["output"]["directory"]]),
+        build["distribution"],
+        "/dev/null",
+        args.mirror,
+        *build.get("extrarepositories", ()),
+    ]
+    proc = subprocess.Popen(cmd,
+                            stdout=None if enablelog else subprocess.DEVNULL,
+                            stderr=subprocess.STDOUT if enablelog
+                                   else subprocess.DEVNULL)
+    sys.exit(proc.wait())
 
 if __name__ == "__main__":
     main()
-- 
cgit v1.2.3