bootc: "Re-locking": use ostree admin unlock --transient

To keep /usr read-only after DNF is finished with a transient
transaction, we call `ostree admin unlock --transient` to mount the /usr
overlay as read-only by default. Then, we create a private mount
namespace for DNF and its child processes and remount the /usr overlayfs
as read/write in the private mountns.

os.unshare is unfortunately only available in Python >= 3.12, so we have
to call libc.unshare via Python ctypes here and hardcode the CLONE_NEWNS
flag that we need to pass.

dnf/cli/cli.py

@@ -218,40 +218,45 @@ def do_transaction(self, display=()):
                 logger.info(_("{prog} will only download packages, install gpg keys, and check the "
                               "transaction.").format(prog=dnf.util.MAIN_PROG_UPPER))
 
-            is_bootc_transaction = dnf.util._Bootc.is_bootc_host() and \
+            is_bootc_transaction = dnf.util._BootcSystem.is_bootc_system() and \
                 os.path.realpath(self.conf.installroot) == "/" and \
                 not self.conf.downloadonly
 
             # Handle bootc transactions. `--transient` must be specified if
             # /usr is not already writeable.
-            bootc = None
+            bootc_system = None
             if is_bootc_transaction:
                 if self.conf.persistence == "persist":
                     logger.info(_("Persistent transactions aren't supported on bootc systems."))
                     raise CliError(_("Operation aborted."))
                 assert self.conf.persistence in ("auto", "transient")
 
-                bootc = dnf.util._Bootc()
+                bootc_system = dnf.util._BootcSystem()
 
-                if not bootc.is_unlocked():
+                if not bootc_system.is_writable():
                     if self.conf.persistence == "auto":
                         logger.info(_("This bootc system is configured to be read-only. Pass --transient to "
                                       "perform this and subsequent transactions in a transient overlay which "
                                       "will reset when the system reboots."))
                         raise CliError(_("Operation aborted."))
                     assert self.conf.persistence == "transient"
-                    logger.info(_("A transient overlay will be created on /usr that will be discarded on reboot. "
-                                  "Keep in mind that changes to /etc and /var will still persist, and packages "
-                                  "commonly modify these directories."))
-            elif self.conf.persistence == "transient":
-                raise CliError(_("Transient transactions are only supported on bootc systems."))
+                    if not bootc_system.is_unlocked_transient():
+                        # Only tell the user about the transient overlay if
+                        # it's not already in place
+                        logger.info(_("A transient overlay will be created on /usr that will be discarded on reboot. "
+                                      "Keep in mind that changes to /etc and /var will still persist, and packages "
+                                      "commonly modify these directories."))
+            else:
+                # Not a bootc transaction.
+                if self.conf.persistence == "transient":
+                    raise CliError(_("Transient transactions are only supported on bootc systems."))
 
             if self._promptWanted():
                 if self.conf.assumeno or not self.output.userconfirm():
                     raise CliError(_("Operation aborted."))
 
-            if bootc:
-                bootc.unlock_and_prepare()
+            if bootc_system:
+                bootc_system.make_writable()
         else:
             logger.info(_('Nothing to do.'))
             return

dnf/const.py.in

@@ -31,6 +31,7 @@ INSTALLONLYPKGS=['kernel', 'kernel-PAE',
                  'installonlypkg(kernel-module)',
                  'installonlypkg(vm)',
                  'multiversion(kernel)']
+LIBC_SONAME = "libc.so.6"
 LOG='dnf.log'
 LOG_HAWKEY='hawkey.log'
 LOG_LIBREPO='dnf.librepo.log'

dnf/util.py

@@ -25,6 +25,7 @@
 from .pycomp import PY3, basestring
 from dnf.i18n import _, ucd
 import argparse
+import ctypes
 import dnf
 import dnf.callback
 import dnf.const
@@ -642,11 +643,12 @@ def _is_file_pattern_present(specs):
     return False
 
 
-class _Bootc:
+class _BootcSystem:
     usr = "/usr"
+    CLONE_NEWNS = 0x00020000 # defined in linux/include/uapi/linux/sched.h
 
     def __init__(self):
-        if not self.is_bootc_host():
+        if not self.is_bootc_system():
             raise RuntimeError(_("Not running on a bootc system."))
 
         import gi
@@ -664,45 +666,81 @@ def __init__(self):
         assert self._booted_deployment is not None
 
     @staticmethod
-    def is_bootc_host():
+    def is_bootc_system():
         """Returns true is the system is managed as an immutable container, false
         otherwise."""
         ostree_booted = "/run/ostree-booted"
         return os.path.isfile(ostree_booted)
 
+    @classmethod
+    def is_writable(cls):
+        """Returns true if and only if /usr is writable."""
+        return os.access(cls.usr, os.W_OK)
+
     def _get_unlocked_state(self):
         return self._booted_deployment.get_unlocked()
 
-    def is_unlocked(self):
-        return self._get_unlocked_state() != self._OSTree.DeploymentUnlockedState.NONE
+    def is_unlocked_transient(self):
+        """Returns true if and only if the bootc system is unlocked in a
+        transient state, i.e. a overlayfs is mounted as read-only on /usr.
+        Changes can be made to the overlayfs by remounting /usr as
+        read/write in a private mount namespace."""
+        return self._get_unlocked_state() == self._OSTree.DeploymentUnlockedState.TRANSIENT
+
+    @classmethod
+    def _set_up_mountns(cls):
+        # os.unshare is only available in Python >= 3.12
+        libc = ctypes.CDLL(dnf.const.LIBC_SONAME)
+        if libc.unshare(cls.CLONE_NEWNS) != 0:
+            raise OSError("Failed to unshare mount namespace")
+
+        mount_command = ["mount", "--options-source=disable", "-o", "remount,rw", cls.usr]
+        try:
+            completed_process = subprocess.run(mount_command, text=True)
+            completed_process.check_returncode()
+        except FileNotFoundError:
+            raise dnf.exceptions.Error(_("%s: command not found.") % mount_command[0])
+        except subprocess.CalledProcessError:
+            raise dnf.exceptions.Error(_("Failed to mount %s as read/write: %s", cls.usr, completed_process.stderr))
 
-    def unlock_and_prepare(self):
-        """Set up a writeable overlay on bootc systems."""
+    @staticmethod
+    def _unlock():
+        unlock_command = ["ostree", "admin", "unlock", "--transient"]
+        try:
+            completed_process = subprocess.run(unlock_command, text=True)
+            completed_process.check_returncode()
+        except FileNotFoundError:
+            raise dnf.exceptions.Error(_("%s: command not found. Is this a bootc system?") % unlock_command[0])
+        except subprocess.CalledProcessError:
+            raise dnf.exceptions.Error(_("Failed to unlock system: %s", completed_process.stderr))
+
+    def make_writable(self):
+        """Set up a writable overlay on bootc systems."""
 
         bootc_unlocked_state = self._get_unlocked_state()
 
         valid_bootc_unlocked_states = (
             self._OSTree.DeploymentUnlockedState.NONE,
             self._OSTree.DeploymentUnlockedState.DEVELOPMENT,
-            # self._OSTree.DeploymentUnlockedState.TRANSIENT,
-            # self._OSTree.DeploymentUnlockedState.HOTFIX,
+            self._OSTree.DeploymentUnlockedState.TRANSIENT,
+            self._OSTree.DeploymentUnlockedState.HOTFIX,
         )
 
         if bootc_unlocked_state not in valid_bootc_unlocked_states:
             raise ValueError(_("Unhandled bootc unlocked state: %s") % bootc_unlocked_state.value_nick)
 
-        if bootc_unlocked_state == self._OSTree.DeploymentUnlockedState.DEVELOPMENT:
-            # System is already unlocked.
+        if bootc_unlocked_state in (self._OSTree.DeploymentUnlockedState.DEVELOPMENT, self._OSTree.DeploymentUnlockedState.HOTFIX):
+            # System is already unlocked in development mode, and usr is
+            # already mounted read/write.
             pass
         elif bootc_unlocked_state == self._OSTree.DeploymentUnlockedState.NONE:
-            unlock_command = ["ostree", "admin", "unlock"]
-
-            try:
-                completed_process = subprocess.run(unlock_command, text=True)
-                completed_process.check_returncode()
-            except FileNotFoundError:
-                raise dnf.exceptions.Error(_("ostree command not found. Is this a bootc system?"))
-            except subprocess.CalledProcessError:
-                raise dnf.exceptions.Error(_("Failed to unlock system: %s", completed_process.stderr))
+            # System is not unlocked. Unlock it in transient mode, then set up
+            # a mount namespace for DNF.
+            self._unlock()
+            self._set_up_mountns()
+        elif bootc_unlocked_state == self._OSTree.DeploymentUnlockedState.TRANSIENT:
+            # System is unlocked in transient mode, so usr is mounted
+            # read-only. Set up a mount namespace for DNF.
+            self._set_up_mountns()
 
         assert os.access(self.usr, os.W_OK)