From 986fb26e3adc4e486a1df52bc7c5c03cc00c95c3 Mon Sep 17 00:00:00 2001 From: Ralf Jung Date: Mon, 15 Apr 2024 15:41:38 +0200 Subject: [PATCH 1/1] refactor the directives system, so that dbus can be entirely implemented without support from the 'core' --- bubblebox.py | 120 ++++++++++++++++++++++++++++++--------------------- profiles.py | 4 +- 2 files changed, 74 insertions(+), 50 deletions(-) diff --git a/bubblebox.py b/bubblebox.py index cb7f514..02bcb5a 100644 --- a/bubblebox.py +++ b/bubblebox.py @@ -6,71 +6,95 @@ XDG_RUNTIME_DIR = os.environ["XDG_RUNTIME_DIR"] BUBBLEBOX_DIR = XDG_RUNTIME_DIR + "/bubblebox" os.makedirs(BUBBLEBOX_DIR, exist_ok=True) -def flat_map(f, xs): - """Concatenate the result of applying `f` to each element of `xs` to a list. - `None` is treated like the empty list.""" - ys = [] - for x in xs: - x_mapped = f(x) - if x_mapped is not None: - ys.extend(x_mapped) - return ys - def randname(): # choose from all lowercase letter letters = string.ascii_lowercase return ''.join(random.choice(letters) for i in range(8)) -class BoxFlags: - """Flags that configure the bubblebox""" - def __init__(self, bwrap_flags = None, dbus_proxy_flags = None): +class BwrapInvocation: + """Gathered information for a bwrap invocation. + This will be created empty, and then each directive's `setup` function is called + with this object, so they can accumulate the bwrap flags and any other relevant state.""" + def __init__(self): + # The flags to pass to bwrap. + self.flags = [] + # Functions to call at the end of the setup process. + # They will receive this object as argument, so they can add further flags. + self.finalizers = [] + # If this is `None` it means so far no d-bus proxy has been set up. + self.dbus_proxy_flags = None + +class BwrapDirective: + """Directive that just passes flags to bwrap.""" + def __init__(self, bwrap_flags): self.bwrap_flags = bwrap_flags - self.dbus_proxy_flags = dbus_proxy_flags + def setup(self, bwrap): + bwrap.flags.extend(self.bwrap_flags) -def launch_dbus_proxy(flags): - """Launches the dbus proxy and returns the bwrap flags to be used to talk to it.""" - # Prepare a pipe to coordinate shutdown of bwrap and the proxy - bwrap_end, other_end = os.pipe() # both FDs are "non-inheritable" now - # Invoke the debus-proxy - filename = BUBBLEBOX_DIR + "/bus-" + randname() - args = ["/usr/bin/xdg-dbus-proxy", "--fd="+str(other_end)] - args += [os.environ["DBUS_SESSION_BUS_ADDRESS"], filename, "--filter"] + flags - #pprint(args) - subprocess.Popen( - args, - pass_fds = [other_end], # default is to pass only the std FDs! - ) - # Wait until the proxy is ready - os.read(bwrap_end, 1) - assert os.path.exists(filename) - # Make sure bwrap can access the other end of the pipe - os.set_inheritable(bwrap_end, True) - # Put this at the usual location for the bus insode the sandbox. - # TODO: What if DBUS_SESSION_BUS_ADDRESS says something else? - return ["--bind", filename, XDG_RUNTIME_DIR + "/bus", "--sync-fd", str(bwrap_end)] +class GroupDirective: + """Directive that groups a bunch of directives to be treated as one.""" + def __init__(self, directives): + self.directives = directives + def setup(self, bwrap): + for directive in self.directives: + directive.setup(bwrap) + +class DbusProxyDirective: + """Directive that sets up a d-bus proxy and adds flags to it. + If the directive is used multiple times, the flags accumulate.""" + def __init__(self, dbus_proxy_flags): + self.dbus_proxy_flags = dbus_proxy_flags + def setup(self, bwrap): + if bwrap.dbus_proxy_flags is None: + # We are the first d-bus proxy directive. Set up the flags and the finalizer. + bwrap.dbus_proxy_flags = [] + bwrap.finalizers.append(DbusProxyDirective.launch_dbus_proxy) + # Always add the flags. + bwrap.dbus_proxy_flags.extend(self.dbus_proxy_flags) + def launch_dbus_proxy(bwrap): + """Finalizer that launches a d-bus proxy with the flags accumulated in `bwrap`.""" + # Prepare a pipe to coordinate shutdown of bwrap and the proxy + bwrap_end, other_end = os.pipe() # both FDs are "non-inheritable" now + # Invoke the debus-proxy + filename = BUBBLEBOX_DIR + "/bus-" + randname() + args = ["/usr/bin/xdg-dbus-proxy", "--fd="+str(other_end)] + args += [os.environ["DBUS_SESSION_BUS_ADDRESS"], filename, "--filter"] + bwrap.dbus_proxy_flags + #pprint(args) + subprocess.Popen( + args, + pass_fds = [other_end], # default is to pass only the std FDs! + ) + # Wait until the proxy is ready + os.read(bwrap_end, 1) + assert os.path.exists(filename) + # Make sure bwrap can access the other end of the pipe + os.set_inheritable(bwrap_end, True) + # Put this at the usual location for the bus insode the sandbox. + # TODO: What if DBUS_SESSION_BUS_ADDRESS says something else? + bwrap.flags.extend(["--bind", filename, XDG_RUNTIME_DIR + "/bus", "--sync-fd", str(bwrap_end)]) # Constructors that should be used instead of directly mentioning the class above. def bwrap_flags(*flags): - return BoxFlags(bwrap_flags=flags) + return BwrapDirective(flags) def dbus_proxy_flags(*flags): - return BoxFlags(dbus_proxy_flags=flags) -def collect_flags(*flags): - bwrap_flags = flat_map(lambda x: x.bwrap_flags, flags) - dbus_proxy_flags = flat_map(lambda x: x.dbus_proxy_flags, flags) - return BoxFlags(bwrap_flags, dbus_proxy_flags) + return DbusProxyDirective(flags) +def group(*directives): + return GroupDirective(directives) # Run the application in the bubblebox with the given flags. -def bubblebox(*flags): +def bubblebox(*directives): if len(sys.argv) <= 1: print(f"USAGE: {sys.argv[0]} ") sys.exit(1) # Make sure `--die-with-parent` is always set. - flags = collect_flags(bwrap_flags("--die-with-parent"), *flags) - bwrap = "/usr/bin/bwrap" - extraflags = [] - if flags.dbus_proxy_flags: - extraflags += launch_dbus_proxy(flags.dbus_proxy_flags) - args = [bwrap] + flags.bwrap_flags + extraflags + ["--"] + sys.argv[1:] + directives = group(bwrap_flags("--die-with-parent"), *directives) + # Compute the bwrap invocation by running all the directives. + bwrap = BwrapInvocation() + directives.setup(bwrap) + for finalizer in bwrap.finalizers: + finalizer(bwrap) + # Run bwrap + args = ["/usr/bin/bwrap"] + bwrap.flags + ["--"] + sys.argv[1:] #pprint(args) os.execvp(args[0], args) diff --git a/profiles.py b/profiles.py index cf87845..70e8b81 100644 --- a/profiles.py +++ b/profiles.py @@ -1,7 +1,7 @@ from bubblebox import * # Various default sandbox settings -DEFAULT = collect_flags( +DEFAULT = group( # namespace unsharing # cannot unshare IPC as that breaks some wine applications bwrap_flags("--unshare-user", "--unshare-pid", "--unshare-cgroup"), @@ -28,7 +28,7 @@ DEFAULT = collect_flags( # https://github.com/igo95862/bubblejail is a good source of paths that need allowing. # We do not give access to pipewire, that needs a portal (https://docs.pipewire.org/page_portal.html). def DESKTOP(name): - return collect_flags( + return group( DEFAULT, # Share XDG_RUNTIME_DIR among all instances of this sandbox shared_runtime_dir(name), -- 2.30.2