lift the dbus functionality from the cgroup example

1 files changed, 68 insertions, 0 deletions

diff --git a/linuxnamespaces/systemd/__init__.py b/linuxnamespaces/systemd/__init__.py
new file mode 100644
index 0000000..d8e7f86
--- /dev/null
+++ b/linuxnamespaces/systemd/__init__.py
@@ -0,0 +1,68 @@
+# Copyright 2024 Helmut Grohne <helmut@subdivi.de>
+# SPDX-License-Identifier: GPL-3
+
+"""Communicate with a systemd instance to create e.g. delegated croups."""
+
+import os
+import sys
+import typing
+
+
+async def start_transient_unit(
+    unitname: str,
+    pids: list[int] | None = None,
+    properties: dict[str, typing.Any] | None = None,
+    dbusdriver: typing.Literal["auto", "jeepney", "dbussy"] = "auto",
+) -> None:
+    """Call the StartTransientUnit dbus method on the user manager."""
+    dbus_properties: list[tuple[str, tuple[str, typing.Any]]] = []
+    if pids is None:
+        pids = [os.getpid()]
+    dbus_properties.append(("PIDs", ("au", pids)))
+    for key, value in ({} if properties is None else properties).items():
+        if isinstance(value, bool):
+            dbus_properties.append((key, ("b", value)))
+        elif isinstance(value, str):
+            dbus_properties.append((key, ("s", value)))
+        else:
+            raise ValueError(
+                f"cannot infer dbus type for property {key} value"
+            )
+    if dbusdriver in ("auto", "jeepney"):
+        try:
+            from .jeepney import start_transient_unit as jeepney_impl
+        except ImportError:
+            pass
+        else:
+            return await jeepney_impl(unitname, dbus_properties)
+    if dbusdriver in ("auto", "dbussy"):
+        try:
+            from .dbussy import start_transient_unit as dbussy_impl
+        except ImportError:
+            pass
+        else:
+            return await dbussy_impl(unitname, dbus_properties)
+    raise NotImplementedError("requested dbusdriver not available")
+
+
+def reexec_as_transient_unit(
+    unitname: str | None = None,
+    properties: dict[str, typing.Any] | None = None,
+    argv: list[str] | None = None,
+) -> typing.NoReturn:
+    """Reexecute the current process via systemd-run thus placing it into a new
+    .scope unit. If no argv is given, sys.argv is used.
+    """
+    execargs = ["systemd-run", "--user", "--scope"]
+    if unitname is not None:
+        execargs.append("--unit=" + unitname)
+    if properties:
+        for key, value in properties.items():
+            if isinstance(value, int):
+                value = str(value)
+            elif not isinstance(value, str):
+                raise ValueError(f"cannot format property {key} value")
+            execargs.append(f"--property={key}={value}")
+    execargs.append("--")
+    execargs.extend(sys.argv if argv is None else argv)
+    os.execvp("systemd-run", execargs)