Repository: osbuild/osbuild
Pull request #346 (2020-04-21 09:40:53)

Unify JSON-Socket Communication
This is a follow-up on the previous `JsonComm` module. I now converted all the callsites to the new module.

This depends on the `libdir for all` #344 PR, since it requires osbuild-access from the runners. I suggest to merge #344 first, then rebase this one.


Patch:
diff --git a/assemblers/org.osbuild.qemu b/assemblers/org.osbuild.qemu
index c7b53f2c..d734ac61 100755
--- a/assemblers/org.osbuild.qemu
+++ b/assemblers/org.osbuild.qemu
@@ -3,7 +3,6 @@
 import contextlib
 import json
 import os
-import socket
 import shutil
 import struct
 import subprocess
@@ -685,10 +684,5 @@ def main(tree, output_dir, options, loop_client):
 
 if __name__ == '__main__':
     args = json.load(sys.stdin)
-
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect("/run/osbuild/api/remoteloop")
-        ret = main(args["tree"], args["output_dir"], args["options"], remoteloop.LoopClient(sock))
-
+    ret = main(args["tree"], args["output_dir"], args["options"], remoteloop.LoopClient("/run/osbuild/api/remoteloop"))
     sys.exit(ret)
diff --git a/assemblers/org.osbuild.rawfs b/assemblers/org.osbuild.rawfs
index 3ee0f430..ae07a059 100755
--- a/assemblers/org.osbuild.rawfs
+++ b/assemblers/org.osbuild.rawfs
@@ -3,7 +3,6 @@
 import contextlib
 import json
 import os
-import socket
 import subprocess
 import sys
 import osbuild.remoteloop as remoteloop
@@ -97,10 +96,5 @@ def main(tree, output_dir, options, loop_client):
 
 if __name__ == '__main__':
     args = json.load(sys.stdin)
-
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect("/run/osbuild/api/remoteloop")
-        r = main(args["tree"], args["output_dir"], args["options"], remoteloop.LoopClient(sock))
-
+    r = main(args["tree"], args["output_dir"], args["options"], remoteloop.LoopClient("/run/osbuild/api/remoteloop"))
     sys.exit(r)
diff --git a/osbuild.spec b/osbuild.spec
index 7757aa33..f4d0dff0 100644
--- a/osbuild.spec
+++ b/osbuild.spec
@@ -74,20 +74,23 @@ make man
 %py3_install
 
 mkdir -p %{buildroot}%{pkgdir}/stages
-install -p -m 0755 $(find stages -type f) %{buildroot}%{pkgdir}/stages/
+install -p -m 0755 $(find stages/* -not -path "*/osbuild") %{buildroot}%{pkgdir}/stages/
+ln -s ../osbuild %{buildroot}%{pkgdir}/stages/osbuild
 
 mkdir -p %{buildroot}%{pkgdir}/assemblers
-install -p -m 0755 $(find assemblers -type f) %{buildroot}%{pkgdir}/assemblers/
+install -p -m 0755 $(find assemblers/* -not -path "*/osbuild") %{buildroot}%{pkgdir}/assemblers/
+ln -s ../osbuild %{buildroot}%{pkgdir}/assemblers/osbuild
 
 mkdir -p %{buildroot}%{pkgdir}/runners
-install -p -m 0755 $(find runners -type f -or -type l) %{buildroot}%{pkgdir}/runners
+install -p -m 0755 $(find runners/* -not -path "*/osbuild") %{buildroot}%{pkgdir}/runners
+ln -s ../osbuild %{buildroot}%{pkgdir}/runners/osbuild
 
 mkdir -p %{buildroot}%{pkgdir}/sources
-install -p -m 0755 $(find sources -type f) %{buildroot}%{pkgdir}/sources
+install -p -m 0755 $(find sources/* -not -path "*/osbuild") %{buildroot}%{pkgdir}/sources
+ln -s ../osbuild %{buildroot}%{pkgdir}/sources/osbuild
 
-# mount points for bind mounting the osbuild library
-mkdir -p %{buildroot}%{pkgdir}/stages/osbuild
-mkdir -p %{buildroot}%{pkgdir}/assemblers/osbuild
+# mount point for bind mounting the osbuild library
+mkdir -p %{buildroot}%{pkgdir}/osbuild
 
 # documentation
 mkdir -p %{buildroot}%{_mandir}/man1
diff --git a/osbuild/api.py b/osbuild/api.py
index e1ce9864..6fdb11a2 100644
--- a/osbuild/api.py
+++ b/osbuild/api.py
@@ -1,14 +1,10 @@
-import array
 import asyncio
 import json
 import os
-import socket
 import sys
 import tempfile
 import threading
-
-
-from . import remoteloop
+from .util import jsoncomm
 
 
 class API:
@@ -39,34 +35,32 @@ class API:
         self._output = os.fdopen(fd)
         return out
 
-    def _setup_stdio(self, sock, addr):
+    def _setup_stdio(self, server, addr):
         with self._prepare_input() as stdin, \
              self._prepare_output() as stdout:
             msg = {}
-            fds = array.array("i")
+            fds = []
             fds.append(stdin.fileno())
             msg['stdin'] = 0
             fds.append(stdout.fileno())
             msg['stdout'] = 1
             fds.append(stdout.fileno())
             msg['stderr'] = 2
-            remoteloop.dump_fds(sock, msg, fds, addr=addr)
 
-    def _dispatch(self, sock):
-        msg, addr = sock.recvfrom(1024)
-        args = json.loads(msg)
-        if args["method"] == 'setup-stdio':
-            self._setup_stdio(sock, addr)
+            server.send(msg, fds=fds, destination=addr)
+
+    def _dispatch(self, server):
+        msg, _, addr = server.recv()
+        if msg["method"] == 'setup-stdio':
+            self._setup_stdio(server, addr)
 
     def _run_event_loop(self):
-        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
-        sock.bind(self.socket_address)
-        self.barrier.wait()
-        self.event_loop.add_reader(sock, self._dispatch, sock)
-        asyncio.set_event_loop(self.event_loop)
-        self.event_loop.run_forever()
-        self.event_loop.remove_reader(sock)
-        sock.close()
+        with jsoncomm.Socket.new_server(self.socket_address) as server:
+            self.barrier.wait()
+            self.event_loop.add_reader(server, self._dispatch, server)
+            asyncio.set_event_loop(self.event_loop)
+            self.event_loop.run_forever()
+            self.event_loop.remove_reader(server)
 
     def __enter__(self):
         self.thread.start()
diff --git a/osbuild/buildroot.py b/osbuild/buildroot.py
index 03e8e09e..43b5bd45 100644
--- a/osbuild/buildroot.py
+++ b/osbuild/buildroot.py
@@ -1,4 +1,6 @@
 
+import importlib
+import importlib.util
 import os
 import platform
 import shutil
@@ -17,7 +19,7 @@ class BuildRoot:
         self.api = tempfile.mkdtemp(prefix="osbuild-api-", dir=path)
         self.var = tempfile.mkdtemp(prefix="osbuild-var-", dir=var)
         self.mounts = []
-        self.libdir = libdir or "/usr/lib/osbuild"
+        self.libdir = libdir
         self.runner = runner
 
         self.mount_root(root)
@@ -76,6 +78,8 @@ class BuildRoot:
         Its arguments mean the same as those for subprocess.run().
         """
 
+        nspawn_ro_binds = []
+
         # pylint suggests to epxlicitly pass `check` to subprocess.run()
         check = kwargs.pop("check", False)
 
@@ -86,6 +90,30 @@ class BuildRoot:
             # wants to be able create devices nodes, so allow that
             loopback_allow += "m"
 
+        # make osbuild API-calls accessible to the container
+        nspawn_ro_binds.append(f"{self.api}:/run/osbuild/api")
+
+        # We want to execute our stages and other scripts in the container. So
+        # far, we do not install osbuild as a package in the container, but
+        # provide it from the outside. Therefore, we need to provide `libdir`
+        # via bind-mount. Furthermore, a system-installed `libdir` has the
+        # python packages separate in `site-packages`, so we need to bind-mount
+        # them as well.
+        # In the future, we want to work towards mandating an osbuild package to
+        # be installed in the container, so the build is self-contained and does
+        # not take scripts from the host. However, this requires osbuild
+        # packaged for those containers. Furthermore, we want to keep supporting
+        # the current import-model for testing and development.
+        if self.libdir is not None:
+            # caller-specified `libdir` must be self-contained
+            nspawn_ro_binds.append(f"{self.libdir}:/run/osbuild/lib")
+        else:
+            # system `libdir` requires importing the python module
+            nspawn_ro_binds.append(f"/usr/lib/osbuild:/run/osbuild/lib")
+            modorigin = importlib.util.find_spec('osbuild').origin
+            modpath = os.path.dirname(modorigin)
+            nspawn_ro_binds.append(f"{modpath}:/run/osbuild/lib/osbuild")
+
         return subprocess.run([
             "systemd-nspawn",
             "--quiet",
@@ -94,9 +122,9 @@ class BuildRoot:
             "--link-journal=no",
             f"--property=DeviceAllow=block-loop {loopback_allow}",
             f"--directory={self.root}",
-            f"--bind-ro={self.libdir}:/run/osbuild/lib",
+            *[f"--bind-ro={b}" for b in nspawn_ro_binds],
             *[f"--bind={b}" for b in (binds or [])],
-            *[f"--bind-ro={b}" for b in [f"{self.api}:/run/osbuild/api"] + (readonly_binds or [])],
+            *[f"--bind-ro={b}" for b in (readonly_binds or [])],
             f"/run/osbuild/lib/runners/{self.runner}"
             ] + argv, check=check, **kwargs)
 
diff --git a/osbuild/loop.py b/osbuild/loop.py
index 9e07b3ce..a52e76fb 100644
--- a/osbuild/loop.py
+++ b/osbuild/loop.py
@@ -239,7 +239,7 @@ class Loop:
 
         fcntl.ioctl(self.fd, self.LOOP_SET_DIRECT_IO, dio)
 
-    def mknod(self, dir_fd=None, mode=0o600):
+    def mknod(self, dir_fd, mode=0o600):
         """Create a secondary device node
 
         Create a device node with the correct name, mode, minor and major
@@ -260,14 +260,12 @@ class Loop:
 
         Parameters
         ----------
-        dir_fd : int, optional
-            Target directory file descriptor, or None to use /dev (None is default)
+        dir_fd : int
+            Target directory file descriptor
         mode : int, optional
             Access mode on the created device node (0o600 is default)
         """
 
-        if not dir_fd:
-            dir_fd = os.open("/dev", os.O_DIRECTORY)
         os.mknod(self.devname,
                  mode=(stat.S_IMODE(mode) | stat.S_IFBLK),
                  device=os.makedev(self.LOOP_MAJOR, self.minor),
diff --git a/osbuild/pipeline.py b/osbuild/pipeline.py
index ee6dd361..72109cd6 100644
--- a/osbuild/pipeline.py
+++ b/osbuild/pipeline.py
@@ -1,7 +1,5 @@
 
 import hashlib
-import importlib
-import importlib.util
 import json
 import os
 import subprocess
@@ -89,12 +87,6 @@ class Stage:
             sources_dir = f"{libdir}/sources" if libdir else "/usr/lib/osbuild/sources"
 
             ro_binds = [f"{sources_output}:/run/osbuild/sources"]
-            if not libdir:
-                osbuild_module_path = os.path.dirname(importlib.util.find_spec('osbuild').origin)
-                # This is a temporary workaround, once we have a common way to include osbuild in the
-                # buildroot we should remove this because it includes code from the host in the buildroot thus
-                # violating our effort of reproducibility.
-                ro_binds.append(f"{osbuild_module_path}:/run/osbuild/lib/stages/osbuild")
 
             with API(f"{build_root.api}/osbuild", args, interactive) as api, \
                 sources.SourcesServer(f"{build_root.api}/sources",
@@ -151,13 +143,8 @@ class Assembler:
                 binds.append(f"{output_dir}:/run/osbuild/output")
                 args["output_dir"] = "/run/osbuild/output"
 
-            osbuild_module_path = os.path.dirname(importlib.util.find_spec('osbuild').origin)
             ro_binds = [f"{tree}:/run/osbuild/tree"]
-            if not libdir:
-                # This is a temporary workaround, once we have a common way to include osbuild in the
-                # buildroot we should remove this because it includes code from the host in the buildroot thus
-                # violating our effort of reproducibility.
-                ro_binds.append(f"{osbuild_module_path}:/run/osbuild/lib/assemblers/osbuild")
+
             with remoteloop.LoopServer(f"{build_root.api}/remoteloop"), \
                 API(f"{build_root.api}/osbuild", args, interactive) as api:
                 r = build_root.run(
diff --git a/osbuild/remoteloop.py b/osbuild/remoteloop.py
index c6a9bc1d..b014b5f6 100644
--- a/osbuild/remoteloop.py
+++ b/osbuild/remoteloop.py
@@ -1,12 +1,10 @@
-import array
 import asyncio
 import contextlib
 import errno
-import json
 import os
-import socket
 import threading
 from . import loop
+from .util import jsoncomm
 
 
 __all__ = [
@@ -15,21 +13,6 @@ __all__ = [
 ]
 
 
-def load_fds(sock, msglen):
-    fds = array.array("i")   # Array of ints
-    msg, ancdata, _, addr = sock.recvmsg(msglen, socket.CMSG_LEN(253 * fds.itemsize))
-    for cmsg_level, cmsg_type, cmsg_data in ancdata:
-        if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS):
-            # Append data, ignoring any truncated integers at the end.
-            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
-    return json.loads(msg), list(fds), addr
-
-
-def dump_fds(sock, obj, fds, flags=0, addr=None):
-    ancillary = [(socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", fds))]
-    sock.sendmsg([json.dumps(obj).encode('utf-8')], ancillary, flags, addr)
-
-
 class LoopServer:
     """Server for creating loopback devices
 
@@ -90,8 +73,8 @@ class LoopServer:
         self.devs.append(lo)
         return lo.devname
 
-    def _dispatch(self, sock):
-        args, fds, addr = load_fds(sock, 1024)
+    def _dispatch(self, server):
+        args, fds, addr = server.recv()
 
         fd = fds[args["fd"]]
         dir_fd = fds[args["dir_fd"]]
@@ -99,18 +82,16 @@ class LoopServer:
         sizelimit = args.get("sizelimit")
 
         devname = self._create_device(fd, dir_fd, offset, sizelimit)
-        ret = json.dumps({"devname": devname})
-        sock.sendto(ret.encode('utf-8'), addr)
+        server.send({"devname": devname}, destination=addr)
+        fds.close()
 
     def _run_event_loop(self):
-        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
-        sock.bind(self.socket_address)
-        self.barrier.wait()
-        self.event_loop.add_reader(sock, self._dispatch, sock)
-        asyncio.set_event_loop(self.event_loop)
-        self.event_loop.run_forever()
-        self.event_loop.remove_reader(sock)
-        sock.close()
+        with jsoncomm.Socket.new_server(self.socket_address) as server:
+            self.barrier.wait()
+            self.event_loop.add_reader(server, self._dispatch, server)
+            asyncio.set_event_loop(self.event_loop)
+            self.event_loop.run_forever()
+            self.event_loop.remove_reader(server)
 
     def __enter__(self):
         self.thread.start()
@@ -125,13 +106,19 @@ class LoopServer:
 
 
 class LoopClient:
-    def __init__(self, sock):
-        self.sock = sock
+    client = None
+
+    def __init__(self, connect_to):
+        self.client = jsoncomm.Socket.new_client(connect_to)
+
+    def __del__(self):
+        if self.client is not None:
+            self.client.close()
 
     @contextlib.contextmanager
     def device(self, filename, offset=None, sizelimit=None):
         req = {}
-        fds = array.array("i")
+        fds = []
 
         fd = os.open(filename, os.O_RDWR)
         dir_fd = os.open("/dev", os.O_DIRECTORY)
@@ -146,12 +133,12 @@ class LoopClient:
         if sizelimit:
             req["sizelimit"] = sizelimit
 
-        dump_fds(self.sock, req, fds)
+        self.client.send(req, fds=fds)
         os.close(dir_fd)
         os.close(fd)
 
-        ret = json.loads(self.sock.recv(1024))
-        path = os.path.join("/dev", ret["devname"])
+        payload, _, _ = self.client.recv()
+        path = os.path.join("/dev", payload["devname"])
         try:
             yield path
         finally:
diff --git a/osbuild/sources.py b/osbuild/sources.py
index 38eeef51..9e9cd574 100644
--- a/osbuild/sources.py
+++ b/osbuild/sources.py
@@ -1,8 +1,8 @@
 import asyncio
 import json
-import socket
 import subprocess
 import threading
+from .util import jsoncomm
 
 
 class SourcesServer:
@@ -39,22 +39,18 @@ class SourcesServer:
         except ValueError:
             return {"error": f"source returned malformed json: {r.stdout}"}
 
-    def _dispatch(self, sock):
-        msg, addr = sock.recvfrom(65536)
-        request = json.loads(msg)
+    def _dispatch(self, server):
+        request, _, addr = server.recv()
         reply = self._run_source(request["source"], request["checksums"])
-        msg = json.dumps(reply).encode("utf-8")
-        sock.sendmsg([msg], [], 0, addr)
+        server.send(reply, destination=addr)
 
     def _run_event_loop(self):
-        sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
-        sock.bind(self.socket_address)
-        self.barrier.wait()
-        self.event_loop.add_reader(sock, self._dispatch, sock)
-        asyncio.set_event_loop(self.event_loop)
-        self.event_loop.run_forever()
-        self.event_loop.remove_reader(sock)
-        sock.close()
+        with jsoncomm.Socket.new_server(self.socket_address) as server:
+            self.barrier.wait()
+            self.event_loop.add_reader(server, self._dispatch, server)
+            asyncio.set_event_loop(self.event_loop)
+            self.event_loop.run_forever()
+            self.event_loop.remove_reader(server)
 
     def __enter__(self):
         self.thread.start()
@@ -67,15 +63,13 @@ class SourcesServer:
 
 
 def get(source, checksums, api_path="/run/osbuild/api/sources"):
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect(api_path)
+    with jsoncomm.Socket.new_client(api_path) as client:
         msg = {
             "source": source,
             "checksums": checksums
         }
-        sock.sendall(json.dumps(msg).encode('utf-8'))
-        reply = json.loads(sock.recv(65536))
+        client.send(msg)
+        reply, _, _ = client.recv()
         if "error" in reply:
             raise RuntimeError(f"{source}: " + reply["error"])
         return reply
diff --git a/osbuild/util/jsoncomm.py b/osbuild/util/jsoncomm.py
new file mode 100644
index 00000000..1591642b
--- /dev/null
+++ b/osbuild/util/jsoncomm.py
@@ -0,0 +1,306 @@
+"""JSON Communication
+
+This module implements a client/server communication method based on JSON
+serialization. It uses unix-domain-datagram-sockets and provides a simple
+unicast message transmission.
+"""
+
+
+import array
+import contextlib
+import errno
+import json
+import os
+import socket
+from typing import Any
+from typing import Optional
+
+
+class FdSet():
+    """File-Descriptor Set
+
+    This object wraps an array of file-descriptors. Unlike a normal integer
+    array, this object owns the file-descriptors and therefore closes them once
+    the object is released.
+
+    File-descriptor sets are initialized once. From then one, the only allowed
+    operation is to query it for information, or steal file-descriptors from
+    it. If you close a set, all remaining file-descriptors are closed and
+    removed from the set. It will then be an empty set.
+    """
+
+    _fds = array.array("i")
+
+    def __init__(self, *, rawfds):
+        for i in rawfds:
+            if not isinstance(i, int) or i < 0:
+                raise ValueError()
+
+        self._fds = rawfds
+
+    def __del__(self):
+        self.close()
+
+    def close(self):
+        """Close All Entries
+
+        This closes all stored file-descriptors and clears the set. Once this
+        returns, the set will be empty. It is safe to call this multiple times.
+        Note that a set is automatically closed when it is garbage collected.
+        """
+
+        for i in self._fds:
+            if i >= 0:
+                os.close(i)
+
+        self._fds = array.array("i")
+
+    @classmethod
+    def from_list(cls, l: list):
+        """Create new Set from List
+
+        This creates a new file-descriptor set initialized to the same entries
+        as in the given list. This consumes the file-descriptors. The caller
+        must not assume ownership anymore.
+        """
+
+        fds = array.array("i")
+        fds.fromlist(l)
+        return cls(rawfds=fds)
+
+    def __len__(self):
+        return len(self._fds)
+
+    def __getitem__(self, key: Any):
+        if self._fds[key] < 0:
+            raise IndexError
+        return self._fds[key]
+
+    def steal(self, key: Any):
+        """Steal Entry
+
+        Retrieve the entry at the given position, but drop it from the internal
+        file-descriptor set. The caller will now own the file-descriptor and it
+        can no longer be accessed through the set.
+
+        Note that this does not reshuffle the set. All indices stay constant.
+        """
+
+        v = self[key]
+        self._fds[key] = -1
+        return v
+
+
+class Socket(contextlib.AbstractContextManager):
+    """Communication Socket
+
+    This socket object represents a communication channel. It allows sending
+    and receiving JSON-encoded messages. It uses unix-domain-datagram sockets
+    as underlying transport.
+    """
+
+    _socket = None
+    _unlink = None
+
+    def __init__(self, sock, unlink):
+        self._socket = sock
+        self._unlink = unlink
+
+    def __del__(self):
+        self.close()
+
+    def __exit__(self, exc_type, exc_value, exc_tb):
+        self.close()
+        return False
+
+    def close(self):
+        """Close Socket
+
+        Close the socket and all underlying resources. This can be called
+        multiple times.
+        """
+
+        # close the socket if it is set
+        if self._socket is not None:
+            self._socket.close()
+            self._socket = None
+
+        # unlink the file-system entry, if pinned
+        if self._unlink is not None:
+            try:
+                os.unlink(self._unlink[1], dir_fd=self._unlink[0])
+            except OSError as e:
+                if e.errno != errno.ENOENT:
+                    raise
+
+            os.close(self._unlink[0])
+            self._unlink = None
+
+    @classmethod
+    def new_client(cls, connect_to: Optional[str] = None):
+        """Create Client
+
+        Create a new client socket.
+
+        Parameters
+        ----------
+        connect_to
+            If not `None`, the client will use the specified address as the
+            default destination for all send operations.
+        """
+
+        sock = None
+
+        try:
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+
+            # Trigger an auto-bind. If you do not do this, you might end up with
+            # an unbound unix socket, which cannot receive messages.
+            # Alternatively, you can also set `SO_PASSCRED`, but this has
+            # side-effects.
+            sock.bind("")
+
+            # Connect the socket. This has no effect other than specifying the
+            # default destination for send operations.
+            if connect_to is not None:
+                sock.connect(connect_to)
+        except:
+            if sock is not None:
+                sock.close()
+            raise
+
+        return cls(sock, None)
+
+    @classmethod
+    def new_server(cls, bind_to: str):
+        """Create Server
+
+        Create a new listener socket.
+
+        Parameters
+        ----------
+        bind_to
+            The socket-address to listen on for incoming client requests.
+        """
+
+        sock = None
+        unlink = None
+        path = os.path.split(bind_to)
+
+        try:
+            # We bind the socket and then open a directory-fd on the target
+            # socket. This allows us to properly unlink the socket when the
+            # server is closed. Note that sockets are never automatically
+            # cleaned up on linux, nor can you bind to existing sockets.
+            # We use a dirfd to guarantee this works even when you change
+            # your mount points in-between.
+            # Yeah, this is racy when mount-points change between the socket
+            # creation and open. But then your entire socket creation is racy
+            # as well. We do not guarantee atomicity, so you better make sure
+            # you do not rely on it.
+            sock = socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM)
+            sock.bind(bind_to)
+            unlink = os.open(os.path.join(".", path[0]), os.O_CLOEXEC | os.O_PATH)
+        except:
+            if unlink is not None:
+                os.close(unlink)
+            if sock is not None:
+                sock.close()
+            raise
+
+        return cls(sock, (unlink, path[1]))
+
+    def fileno(self) -> int:
+        assert self._socket is not None
+        return self._socket.fileno()
+
+    def recv(self):
+        """Receive a Message
+
+        This receives the next pending message from the socket. This operation
+        is synchronous.
+
+        A tuple consisting of the deserialized message payload, the auxiliary
+        file-descriptor set, and the socket-address of the sender is returned.
+        """
+
+        # On `SOCK_DGRAM`, packets might be arbitrarily sized. There is no
+        # hard-coded upper limit, since it is only restricted by the size of
+        # the kernel write buffer on sockets (which itself can be modified via
+        # sysctl). The only real maximum is probably something like 2^31-1,
+        # since that is the maximum of that sysctl datatype.
+        # Anyway, `MSG_TRUNC+MSG_PEEK` usually allows us to easily peek at the
+        # incoming buffer. Unfortunately, the python `recvmsg()` wrapper
+        # discards the return code and we cannot use that. Instead, we simply
+        # loop until we know the size. This is slightly awkward, but seems fine
+        # as long as you do not put this into a hot-path.
+        size = 4096
+        while True:
+            peek = self._socket.recvmsg(size, 0, socket.MSG_PEEK)
+            if not (peek[2] & socket.MSG_TRUNC):
+                break
+            size *= 2
+
+        # Fetch a packet from the socket. On linux, the maximum SCM_RIGHTS array
+        # size is hard-coded to 253. This allows us to size the ancillary buffer
+        # big enough to receive any possible message.
+        fds = array.array("i")
+        msg = self._socket.recvmsg(size, socket.CMSG_LEN(253 * fds.itemsize))
+
+        # First thing we do is always to fetch the CMSG FDs into an FdSet. This
+        # guarantees that we do not leak FDs in case the message handling fails
+        # for other reasons.
+        for level, ty, data in msg[1]:
+            if level == socket.SOL_SOCKET and ty == socket.SCM_RIGHTS:
+                assert len(data) % fds.itemsize == 0
+                fds.frombytes(data)
+        fdset = FdSet(rawfds=fds)
+
+        # Check the returned message flags. If the message was truncated, we
+        # have to discard it. This shouldn't happen, but there is no harm in
+        # handling it. However, `CTRUNC` can happen, since it is also triggered
+        # when LSMs reject FD transmission. Treat it the same as a parser error.
+        flags = msg[2]
+        if flags & (socket.MSG_TRUNC | socket.MSG_CTRUNC):
+            raise BufferError
+
+        try:
+            payload = json.loads(msg[0])
+        except json.JSONDecodeError:
+            raise BufferError
+
+        return (payload, fdset, msg[3])
+
+    def send(self, payload: object, *, destination: Optional[str] = None, fds: list = []):
+        """Send Message
+
+        Send a new message via this socket. This operation is synchronous. The
+        maximum message size depends on the configured send-buffer on the
+        socket. An `OSError` with `EMSGSIZE` is raised when it is exceeded.
+
+        Parameters
+        ----------
+        payload
+            A python object to serialize as JSON and send via this socket. See
+            `json.dump()` for details about the serialization involved.
+        destination
+            The destination to send to. If `None`, the default destination is
+            used (if none is set, this will raise an `OSError`).
+        fds
+            A list of file-descriptors to send with the message.
+
+        Raises
+        ------
+        OSError
+            If the socket cannot be written, a matching `OSError` is raised.
+        TypeError
+            If the payload cannot be serialized, a type error is raised.
+        """
+
+        serialized = json.dumps(payload).encode()
+        cmsg = []
+        if len(fds) > 0:
+            cmsg.append((socket.SOL_SOCKET, socket.SCM_RIGHTS, array.array("i", fds)))
+
+        n = self._socket.sendmsg([serialized], cmsg, 0, destination)
+        assert n == len(serialized)
diff --git a/runners/org.osbuild.fedora30 b/runners/org.osbuild.fedora30
index 93f1374f..3768dbe9 100755
--- a/runners/org.osbuild.fedora30
+++ b/runners/org.osbuild.fedora30
@@ -1,23 +1,11 @@
 #!/usr/bin/python3
 
-import array
-import json
 import os
-import socket
 import subprocess
 import sys
+from osbuild.util import jsoncomm
 
 
-# copied from remoteloop.py
-def load_fds(sock, msglen):
-    fds = array.array("i")   # Array of ints
-    msg, ancdata, _, addr = sock.recvmsg(msglen, socket.CMSG_LEN(253 * fds.itemsize))
-    for cmsg_level, cmsg_type, cmsg_data in ancdata:
-        if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS):
-            # Append data, ignoring any truncated integers at the end.
-            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
-    return json.loads(msg), list(fds), addr
-
 def ldconfig():
     # ld.so.conf must exist, or `ldconfig` throws a warning
     subprocess.run(["touch", "/etc/ld.so.conf"], check=True)
@@ -49,17 +37,15 @@ def nsswitch():
 
 
 def setup_stdio():
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect("/run/osbuild/api/osbuild")
+    with jsoncomm.Socket.new_client("/run/osbuild/api/osbuild") as client:
         req = {'method': 'setup-stdio'}
-        sock.send(json.dumps(req).encode('utf-8'))
-        msg, fds, _ = load_fds(sock, 1024)
+        client.send(req)
+        msg, fds, _ = client.recv()
         for io in ['stdin', 'stdout', 'stderr']:
             target = getattr(sys, io)
             source = fds[msg[io]]
             os.dup2(source, target.fileno())
-            os.close(source)
+        fds.close()
 
 
 if __name__ == "__main__":
diff --git a/runners/org.osbuild.linux b/runners/org.osbuild.linux
index b58ec583..ffa1c5b1 100755
--- a/runners/org.osbuild.linux
+++ b/runners/org.osbuild.linux
@@ -1,34 +1,21 @@
 #!/usr/bin/python3
 
-import array
-import json
 import os
-import socket
 import subprocess
 import sys
+from osbuild.util import jsoncomm
 
 
-def load_fds(sock, msglen):
-    fds = array.array("i")   # Array of ints
-    msg, ancdata, _, addr = sock.recvmsg(msglen, socket.CMSG_LEN(253 * fds.itemsize))
-    for cmsg_level, cmsg_type, cmsg_data in ancdata:
-        if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS):
-            # Append data, ignoring any truncated integers at the end.
-            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
-    return json.loads(msg), list(fds), addr
-
 def setup_stdio():
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect("/run/osbuild/api/osbuild")
+    with jsoncomm.Socket.new_client("/run/osbuild/api/osbuild") as client:
         req = {'method': 'setup-stdio'}
-        sock.send(json.dumps(req).encode('utf-8'))
-        msg, fds, _ = load_fds(sock, 1024)
+        client.send(req)
+        msg, fds, _ = client.recv()
         for io in ['stdin', 'stdout', 'stderr']:
             target = getattr(sys, io)
             source = fds[msg[io]]
             os.dup2(source, target.fileno())
-            os.close(source)
+        fds.close()
 
 
 if __name__ == "__main__":
diff --git a/runners/org.osbuild.rhel81 b/runners/org.osbuild.rhel81
index a7a95264..6fe2f1d6 100755
--- a/runners/org.osbuild.rhel81
+++ b/runners/org.osbuild.rhel81
@@ -1,23 +1,11 @@
 #!/usr/bin/python3.6
 
-import array
-import json
 import os
-import socket
 import subprocess
 import sys
+from osbuild.util import jsoncomm
 
 
-# copied from remoteloop.py
-def load_fds(sock, msglen):
-    fds = array.array("i")   # Array of ints
-    msg, ancdata, _, addr = sock.recvmsg(msglen, socket.CMSG_LEN(253 * fds.itemsize))
-    for cmsg_level, cmsg_type, cmsg_data in ancdata:
-        if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS):
-            # Append data, ignoring any truncated integers at the end.
-            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
-    return json.loads(msg), list(fds), addr
-
 def ldconfig():
     # ld.so.conf must exist, or `ldconfig` throws a warning
     subprocess.run(["touch", "/etc/ld.so.conf"], check=True)
@@ -49,17 +37,15 @@ def nsswitch():
 
 
 def setup_stdio():
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect("/run/osbuild/api/osbuild")
+    with jsoncomm.Socket.new_client("/run/osbuild/api/osbuild") as client:
         req = {'method': 'setup-stdio'}
-        sock.send(json.dumps(req).encode('utf-8'))
-        msg, fds, _ = load_fds(sock, 1024)
+        client.send(req)
+        msg, fds, _ = client.recv()
         for io in ['stdin', 'stdout', 'stderr']:
             target = getattr(sys, io)
             source = fds[msg[io]]
             os.dup2(source, target.fileno())
-            os.close(source)
+        fds.close()
 
 def os_release():
     """/usr/lib/os-release doesn't exist. The `redhat-release` package
diff --git a/runners/org.osbuild.rhel82 b/runners/org.osbuild.rhel82
index 235e1cdf..160df804 100755
--- a/runners/org.osbuild.rhel82
+++ b/runners/org.osbuild.rhel82
@@ -1,23 +1,11 @@
 #!/usr/bin/python3.6
 
-import array
-import json
 import os
-import socket
 import subprocess
 import sys
+from osbuild.util import jsoncomm
 
 
-# copied from remoteloop.py
-def load_fds(sock, msglen):
-    fds = array.array("i")   # Array of ints
-    msg, ancdata, _, addr = sock.recvmsg(msglen, socket.CMSG_LEN(253 * fds.itemsize))
-    for cmsg_level, cmsg_type, cmsg_data in ancdata:
-        if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS):
-            # Append data, ignoring any truncated integers at the end.
-            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
-    return json.loads(msg), list(fds), addr
-
 def ldconfig():
     # ld.so.conf must exist, or `ldconfig` throws a warning
     subprocess.run(["touch", "/etc/ld.so.conf"], check=True)
@@ -49,17 +37,16 @@ def nsswitch():
 
 
 def setup_stdio():
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect("/run/osbuild/api/osbuild")
+    with jsoncomm.Socket.new_client("/run/osbuild/api/osbuild") as client:
         req = {'method': 'setup-stdio'}
-        sock.send(json.dumps(req).encode('utf-8'))
-        msg, fds, _ = load_fds(sock, 1024)
+        client.send(req)
+        msg, fds, _ = client.recv()
         for io in ['stdin', 'stdout', 'stderr']:
             target = getattr(sys, io)
             source = fds[msg[io]]
             os.dup2(source, target.fileno())
-            os.close(source)
+        fds.close()
+
 
 def python_alternatives():
     """/usr/bin/python3 is a symlink to /etc/alternatives/python3, which points
diff --git a/runners/org.osbuild.rhel83 b/runners/org.osbuild.rhel83
new file mode 120000
index 00000000..fe28a86c
--- /dev/null
+++ b/runners/org.osbuild.rhel83
@@ -0,0 +1,1 @@
+org.osbuild.rhel82
\ No newline at end of file
diff --git a/runners/org.osbuild.ubuntu1804 b/runners/org.osbuild.ubuntu1804
index 93f1374f..3768dbe9 100755
--- a/runners/org.osbuild.ubuntu1804
+++ b/runners/org.osbuild.ubuntu1804
@@ -1,23 +1,11 @@
 #!/usr/bin/python3
 
-import array
-import json
 import os
-import socket
 import subprocess
 import sys
+from osbuild.util import jsoncomm
 
 
-# copied from remoteloop.py
-def load_fds(sock, msglen):
-    fds = array.array("i")   # Array of ints
-    msg, ancdata, _, addr = sock.recvmsg(msglen, socket.CMSG_LEN(253 * fds.itemsize))
-    for cmsg_level, cmsg_type, cmsg_data in ancdata:
-        if (cmsg_level == socket.SOL_SOCKET and cmsg_type == socket.SCM_RIGHTS):
-            # Append data, ignoring any truncated integers at the end.
-            fds.frombytes(cmsg_data[:len(cmsg_data) - (len(cmsg_data) % fds.itemsize)])
-    return json.loads(msg), list(fds), addr
-
 def ldconfig():
     # ld.so.conf must exist, or `ldconfig` throws a warning
     subprocess.run(["touch", "/etc/ld.so.conf"], check=True)
@@ -49,17 +37,15 @@ def nsswitch():
 
 
 def setup_stdio():
-    with socket.socket(socket.AF_UNIX, socket.SOCK_DGRAM) as sock:
-        sock.setsockopt(socket.SOL_SOCKET, socket.SO_PASSCRED, 1)
-        sock.connect("/run/osbuild/api/osbuild")
+    with jsoncomm.Socket.new_client("/run/osbuild/api/osbuild") as client:
         req = {'method': 'setup-stdio'}
-        sock.send(json.dumps(req).encode('utf-8'))
-        msg, fds, _ = load_fds(sock, 1024)
+        client.send(req)
+        msg, fds, _ = client.recv()
         for io in ['stdin', 'stdout', 'stderr']:
             target = getattr(sys, io)
             source = fds[msg[io]]
             os.dup2(source, target.fileno())
-            os.close(source)
+        fds.close()
 
 
 if __name__ == "__main__":
diff --git a/runners/osbuild b/runners/osbuild
new file mode 120000
index 00000000..d41e248d
--- /dev/null
+++ b/runners/osbuild
@@ -0,0 +1,1 @@
+../osbuild
\ No newline at end of file
diff --git a/sources/osbuild b/sources/osbuild
new file mode 120000
index 00000000..d41e248d
--- /dev/null
+++ b/sources/osbuild
@@ -0,0 +1,1 @@
+../osbuild
\ No newline at end of file
diff --git a/stages/org.osbuild.grub2 b/stages/org.osbuild.grub2
index dbb171f6..6af2c0a9 100755
--- a/stages/org.osbuild.grub2
+++ b/stages/org.osbuild.grub2
@@ -254,8 +254,19 @@ def main(tree, options):
 
     with open(grubenv, "w") as env:
         fs_type, fs_id = fs_spec_decode(root_fs)
-        env.write("# GRUB Environment Block\n"
-                  f"kernelopts=root={fs_type}={fs_id} {kernel_opts}\n")
+        data = (
+            "# GRUB Environment Block\n"
+            f"kernelopts=root={fs_type}={fs_id} {kernel_opts}\n"
+        )
+
+        # The 'grubenv' file is, according to the documentation,
+        # a 'preallocated 1024-byte file'. The empty space is
+        # needs to be filled with '#' as padding
+        data += '#' * (1024 - len(data))
+        assert len(data) == 1024
+
+        print(data)
+        env.write(data)
 
     if uefi is not None:
         # UEFI support:


Test patch:
diff --git a/test/test_util_jsoncomm.py b/test/test_util_jsoncomm.py
new file mode 100644
index 00000000..310f26dd
--- /dev/null
+++ b/test/test_util_jsoncomm.py
@@ -0,0 +1,159 @@
+#
+# Tests for the 'osbuild.util.jsoncomm' module.
+#
+
+
+import asyncio
+import os
+import tempfile
+import unittest
+from osbuild.util import jsoncomm
+
+
+class TestUtilJsonComm(unittest.TestCase):
+    def setUp(self):
+        self.dir = tempfile.TemporaryDirectory()
+        self.address = os.path.join(self.dir.name, "listener")
+        self.server = jsoncomm.Socket.new_server(self.address)
+        self.client = jsoncomm.Socket.new_client(self.address)
+
+    def tearDown(self):
+        self.client.close()
+        self.server.close()
+        self.dir.cleanup()
+
+    def test_fdset(self):
+        #
+        # Test the FdSet implementation. Create a simple FD array and verify
+        # that the FdSet correctly indexes them. Furthermore, verify that a
+        # close actually closes the Fds so a following FdSet will get the same
+        # FD numbers assigned.
+        #
+
+        v1 = [os.dup(0), os.dup(0), os.dup(0), os.dup(0)]
+        s = jsoncomm.FdSet.from_list(v1)
+        assert len(s) == 4
+        for i in range(4):
+            assert s[i] == v1[i]
+        with self.assertRaises(IndexError):
+            _ = s[128]
+        s.close()
+
+        v2 = [os.dup(0), os.dup(0), os.dup(0), os.dup(0)]
+        assert v1 == v2
+        s = jsoncomm.FdSet.from_list(v2)
+        s.close()
+
+    def test_fdset_init(self):
+        #
+        # Test FdSet initializations. This includes common edge-cases like empty
+        # initializers, invalid array values, or invalid types.
+        #
+
+        s = jsoncomm.FdSet.from_list([])
+        s.close()
+
+        with self.assertRaises(ValueError):
+            v1 = [-1]
+            s = jsoncomm.FdSet.from_list(v1)
+
+        with self.assertRaises(ValueError):
+            v1 = ["foobar"]
+            s = jsoncomm.FdSet(rawfds=v1)
+
+    def test_ping_pong(self):
+        #
+        # Test sending messages through the client/server connection.
+        #
+
+        data = {"key": "value"}
+        self.client.send(data)
+        msg = self.server.recv()
+        assert msg[0] == data
+        assert len(msg[1]) == 0
+
+        self.server.send(data, destination=msg[2])
+        msg = self.client.recv()
+        assert msg[0] == data
+        assert len(msg[1]) == 0
+
+    def test_scm_rights(self):
+        #
+        # Test FD transmission. Create a file, send a file-descriptor through
+        # the communication channel, and then verify that the file-contents
+        # can be read.
+        #
+
+        with tempfile.TemporaryFile() as f1:
+            f1.write(b"foobar")
+            f1.seek(0)
+
+            self.client.send({}, fds=[f1.fileno()])
+
+            msg = self.server.recv()
+            assert msg[0] == {}
+            assert len(msg[1]) == 1
+            with os.fdopen(msg[1].steal(0)) as f2:
+                assert f2.read() == "foobar"
+
+    def test_listener_cleanup(self):
+        #
+        # Verify that only a single server can listen on a specified address.
+        # Then make sure closing a server will correctly unlink its socket.
+        #
+
+        addr = os.path.join(self.dir.name, "foobar")
+        srv1 = jsoncomm.Socket.new_server(addr)
+        with self.assertRaises(OSError):
+            srv2 = jsoncomm.Socket.new_server(addr)
+        srv1.close()
+        srv2 = jsoncomm.Socket.new_server(addr)
+        srv2.close()
+
+    def test_contextlib(self):
+        #
+        # Verify the context-manager of sockets. Make sure they correctly close
+        # the socket, and they correctly propagate exceptions.
+        #
+
+        assert self.client.fileno() >= 0
+        with self.client as client:
+            assert client == self.client
+            assert client.fileno() >= 0
+        with self.assertRaises(AssertionError):
+            self.client.fileno()
+
+        assert self.server.fileno() >= 0
+        with self.assertRaises(SystemError):
+            with self.server as server:
+                assert server.fileno() >= 0
+                raise SystemError
+            raise AssertionError
+        with self.assertRaises(AssertionError):
+            self.server.fileno()
+
+    def test_asyncio(self):
+        #
+        # Test integration with asyncio-eventloops. Use a trivial echo server
+        # and test a simple ping/pong roundtrip.
+        #
+
+        loop = asyncio.new_event_loop()
+
+        def echo(socket):
+            msg = socket.recv()
+            socket.send(msg[0], destination=msg[2])
+            loop.stop()
+
+        self.client.send({})
+
+        loop.add_reader(self.server, echo, self.server)
+        loop.run_forever()
+        loop.close()
+
+        msg = self.client.recv()
+        assert msg[0] == {}
+
+
+if __name__ == "__main__":
+    unittest.main()