Merge pull request #1135 from vojtechtrefny/3.8-devel_libblockdev-30

Prepare for libblockdev 3.0 release

.github/workflows/check.yml

@@ -14,15 +14,36 @@ on:
 
 jobs:
   build:
-
+    name: static-analysis
     runs-on: ubuntu-22.04
-
+    env:
+      CI_IMAGE: fedora:latest
+      CI_CONTAINER: blivet-tests
     steps:
-    - uses: actions/checkout@v3
-    - name: Install dependencies
-      run:   |
-         sudo apt-get -qq update
-         sudo apt-get -y -qq install make ansible
-         sudo make install-requires
-    - name: Run checks
-      run: make check
+      - name: Checkout libblockdev repository
+        uses: actions/checkout@v3
+
+      - name: Install podman
+        run: |
+          sudo apt -qq update
+          sudo apt -y -qq install podman
+
+      - name: Start the container
+        run: |
+          podman run -d -t --name ${{ env.CI_CONTAINER }} --privileged --volume "$(pwd):/app" --workdir "/app" ${{ env.CI_IMAGE }}
+
+      - name: Install ansible in the container
+        run: |
+          podman exec -it ${{ env.CI_CONTAINER }} bash -c "dnf -y install ansible make which"
+
+      - name: Enable our daily builds Copr in the container
+        run: |
+          podman exec -it ${{ env.CI_CONTAINER }} bash -c "dnf -y install dnf-plugins-core && dnf -y copr enable @storage/udisks-daily"
+
+      - name: Install test dependencies in the container
+        run: |
+          podman exec -it ${{ env.CI_CONTAINER }} bash -c "ansible-playbook -i 'localhost,' -c local misc/install-test-dependencies.yml"
+
+      - name: Run static analysis tests in the container
+        run: |
+          podman exec -it ${{ env.CI_CONTAINER }} bash -c "make check"

blivet/__init__.py

@@ -73,9 +73,6 @@ def log_bd_message(level, msg):
 
 _requested_plugins = blockdev.plugin_specs_from_names(_REQUESTED_PLUGIN_NAMES)
 try:
-    # do not check for dependencies during libblockdev initialization, do runtime
-    # checks instead
-    blockdev.switch_init_checks(False)
     succ_, avail_plugs = blockdev.try_reinit(require_plugins=_requested_plugins, reload=False, log_func=log_bd_message)
 except GLib.GError as err:
     raise RuntimeError("Failed to initialize the libblockdev library: %s" % err)

blivet/formats/luks.py

@@ -262,10 +262,17 @@ def _pre_setup(self, **kwargs):
     def _setup(self, **kwargs):
         log_method_call(self, device=self.device, map_name=self.map_name,
                         type=self.type, status=self.status)
+
+        # passphrase is preferred for open
+        if self.__passphrase:
+            context = blockdev.CryptoKeyslotContext(passphrase=self.__passphrase)
+        elif self._key_file:
+            context = blockdev.CryptoKeyslotContext(keyfile=self._key_file)
+        else:
+            raise LUKSError("Passphrase or key file must be set for LUKS setup")
+
         try:
-            blockdev.crypto.luks_open(self.device, self.map_name,
-                                      passphrase=self.__passphrase,
-                                      key_file=self._key_file)
+            blockdev.crypto.luks_open(self.device, self.map_name, context=context)
         except blockdev.CryptoError as e:
             raise LUKSError(e)
 
@@ -328,10 +335,16 @@ def _create(self, **kwargs):
             else:
                 extra = None
 
+        if self.__passphrase:
+            context = blockdev.CryptoKeyslotContext(passphrase=self.__passphrase)
+        elif self._key_file:
+            context = blockdev.CryptoKeyslotContext(keyfile=self._key_file)
+        else:
+            raise LUKSError("Passphrase or key file must be set for LUKS create")
+
         try:
             blockdev.crypto.luks_format(self.device,
-                                        passphrase=self.__passphrase,
-                                        key_file=self._key_file,
+                                        context=context,
                                         cipher=self.cipher,
                                         key_size=self.key_size,
                                         min_entropy=self.min_luks_entropy,
@@ -340,13 +353,23 @@ def _create(self, **kwargs):
         except blockdev.CryptoError as e:
             raise LUKSError(e)
 
+        if self.__passphrase and self._key_file:
+            # both passphrase and keyfile are set, we need to add the keyfile too
+            ncontext = blockdev.CryptoKeyslotContext(keyfile=self._key_file)
+            try:
+                blockdev.crypto.luks_add_key(self.device, context, ncontext)
+            except blockdev.CryptoError as e:
+                raise LUKSError(e)
+
     def _post_create(self, **kwargs):
         super(LUKS, self)._post_create(**kwargs)
 
         try:
-            self.uuid = blockdev.crypto.luks_uuid(self.device)
+            info = blockdev.crypto.luks_info(self.device)
         except blockdev.CryptoError as e:
             raise LUKSError("Failed to get UUID for the newly created LUKS device %s: %s" % (self.device, str(e)))
+        else:
+            self.uuid = info.uuid
 
         if not self.map_name:
             self.map_name = "luks-%s" % self.uuid
@@ -371,11 +394,17 @@ def add_passphrase(self, passphrase):
         if not self.exists:
             raise LUKSError("format has not been created")
 
+        # if both passphrase and keyfile are set, they both need to be valid so
+        # we can choose passphrase as default
+        if self.__passphrase:
+            context = blockdev.CryptoKeyslotContext(passphrase=self.__passphrase)
+        elif self._key_file:
+            context = blockdev.CryptoKeyslotContext(keyfile=self._key_file)
+
+        ncontext = blockdev.CryptoKeyslotContext(passphrase=passphrase)
+
         try:
-            blockdev.crypto.luks_add_key(self.device,
-                                         pass_=self.__passphrase,
-                                         key_file=self._key_file,
-                                         npass=passphrase)
+            blockdev.crypto.luks_add_key(self.device, context, ncontext)
         except blockdev.CryptoError as e:
             raise LUKSError(e)
 
@@ -384,19 +413,27 @@ def remove_passphrase(self):
         Remove the saved passphrase (and possibly key file) from the LUKS
         header.
 
+        Note: If both passphrase and keyfile are set for this format, both
+              will be removed!
         """
 
         log_method_call(self, device=self.device,
                         type=self.type, status=self.status)
         if not self.exists:
             raise LUKSError("format has not been created")
 
-        try:
-            blockdev.crypto.luks_remove_key(self.device,
-                                            pass_=self.__passphrase,
-                                            key_file=self._key_file)
-        except blockdev.CryptoError as e:
-            raise LUKSError(e)
+        def _remove_passphrase(context):
+            try:
+                blockdev.crypto.luks_remove_key(self.device, context=context)
+            except blockdev.CryptoError as e:
+                raise LUKSError(e)
+
+        if self.__passphrase:
+            context = blockdev.CryptoKeyslotContext(passphrase=self.__passphrase)
+            _remove_passphrase(context)
+        elif self._key_file:
+            context = blockdev.CryptoKeyslotContext(keyfile=self._key_file)
+            _remove_passphrase(context)
 
     def escrow(self, directory, backup_passphrase):
         log.debug("escrow: escrow_volume start for %s", self.device)

blivet/populator/helpers/loop.py

@@ -39,12 +39,15 @@ def run(self):
         name = udev.device_get_name(self.data)
         log_method_call(self, name=name)
         sysfs_path = udev.device_get_sysfs_path(self.data)
-        backing_file = blockdev.loop.get_backing_file(name)
-        if backing_file is None:
+        try:
+            info = blockdev.loop.info(name)
+        except blockdev.LoopError:
             return None
-        file_device = self._devicetree.get_device_by_name(backing_file)
+        if not info.backing_file:
+            return None
+        file_device = self._devicetree.get_device_by_name(info.backing_file)
         if not file_device:
-            file_device = FileDevice(backing_file, exists=True)
+            file_device = FileDevice(info.backing_file, exists=True)
             self._devicetree._add_device(file_device)
         device = LoopDevice(name,
                             parents=[file_device],

blivet/static_data/nvdimm.py

@@ -34,7 +34,7 @@ class NVDIMMDependencyGuard(util.DependencyGuard):
 
     def _check_avail(self):
         try:
-            BlockDev.nvdimm_is_tech_avail(BlockDev.NVDIMMTech.NVDIMM_TECH_NAMESPACE,
+            BlockDev.nvdimm_is_tech_avail(BlockDev.NVDIMMTech.NAMESPACE,
                                           BlockDev.NVDIMMTechMode.RECONFIGURE |
                                           BlockDev.NVDIMMTechMode.QUERY |
                                           BlockDev.NVDIMMTechMode.ACTIVATE_DEACTIVATE)

blivet/tasks/availability.py

@@ -371,7 +371,6 @@ def available_resource(name):
 BLOCKDEV_CRYPTO = BlockDevTechInfo(plugin_name="crypto",
                                    check_fn=blockdev.crypto_is_tech_avail,
                                    technologies={blockdev.CryptoTech.LUKS: BLOCKDEV_CRYPTO_ALL_MODES,
-                                                 blockdev.CryptoTech.LUKS2: BLOCKDEV_CRYPTO_ALL_MODES,
                                                  blockdev.CryptoTech.ESCROW: blockdev.CryptoTechMode.CREATE})
 BLOCKDEV_CRYPTO_TECH = BlockDevMethod(BLOCKDEV_CRYPTO)
 
@@ -399,7 +398,7 @@ def available_resource(name):
                            blockdev.LoopTechMode.QUERY)
 BLOCKDEV_LOOP = BlockDevTechInfo(plugin_name="loop",
                                  check_fn=blockdev.loop_is_tech_avail,
-                                 technologies={blockdev.LoopTech.LOOP_TECH_LOOP: BLOCKDEV_LOOP_ALL_MODES})
+                                 technologies={blockdev.LoopTech.LOOP: BLOCKDEV_LOOP_ALL_MODES})
 BLOCKDEV_LOOP_TECH = BlockDevMethod(BLOCKDEV_LOOP)
 
 # libblockdev lvm plugin required technologies and modes
@@ -437,7 +436,7 @@ def available_resource(name):
                          blockdev.MDTechMode.QUERY)
 BLOCKDEV_MD = BlockDevTechInfo(plugin_name="mdraid",
                                check_fn=blockdev.md_is_tech_avail,
-                               technologies={blockdev.MDTech.MD_TECH_MDRAID: BLOCKDEV_MD_ALL_MODES})
+                               technologies={blockdev.MDTech.MDRAID: BLOCKDEV_MD_ALL_MODES})
 BLOCKDEV_MD_TECH = BlockDevMethod(BLOCKDEV_MD)
 
 # libblockdev mpath plugin required technologies and modes
@@ -455,7 +454,7 @@ def available_resource(name):
                            blockdev.SwapTechMode.SET_LABEL)
 BLOCKDEV_SWAP = BlockDevTechInfo(plugin_name="swap",
                                  check_fn=blockdev.swap_is_tech_avail,
-                                 technologies={blockdev.SwapTech.SWAP_TECH_SWAP: BLOCKDEV_SWAP_ALL_MODES})
+                                 technologies={blockdev.SwapTech.SWAP: BLOCKDEV_SWAP_ALL_MODES})
 BLOCKDEV_SWAP_TECH = BlockDevMethod(BLOCKDEV_SWAP)
 
 # libblockdev plugins

blivet/tasks/lukstasks.py

@@ -87,9 +87,15 @@ def do_task(self):  # pylint: disable=arguments-differ
         """ Resizes the LUKS format. """
         try:
             if self.luks.luks_version == "luks2":
+                if self.luks._LUKS__passphrase:
+                    context = blockdev.CryptoKeyslotContext(passphrase=self.luks._LUKS__passphrase)
+                elif self.luks._key_file:
+                    context = blockdev.CryptoKeyslotContext(keyfile=self.luks._key_file)
+                else:
+                    # context for resize can be NULL -- this means the key is already in the keyring
+                    context = None
                 blockdev.crypto.luks_resize(self.luks.map_name, self.luks.target_size.convert_to(self.unit),
-                                            passphrase=self.luks._LUKS__passphrase,
-                                            key_file=self.luks._key_file)
+                                            context=context)
             else:
                 blockdev.crypto.luks_resize(self.luks.map_name, self.luks.target_size.convert_to(self.unit))
         except blockdev.CryptoError as e:

blivet/util.py

@@ -294,7 +294,12 @@ def get_mount_device(mountpoint):
 
     if mount_device and re.match(r'/dev/loop\d+$', mount_device):
         loop_name = os.path.basename(mount_device)
-        mount_device = blockdev.loop.get_backing_file(loop_name)
+        try:
+            info = blockdev.loop.info(loop_name)
+        except blockdev.LoopError as e:
+            log.warning("failed to get loop info for %s: %s", loop_name, str(e))
+            return None
+        mount_device = info.backing_file
         log.debug("found backing file %s for loop device %s", mount_device,
                   loop_name)
 

tests/storage_tests/formats_test/luks_test.py

@@ -1,8 +1,10 @@
 import os
+import tempfile
 import unittest
 
 from blivet.formats.luks import LUKS, Integrity
 from blivet.devicelibs import crypto
+from blivet.errors import LUKSError
 from blivet.size import Size
 
 from . import loopbackedtestcase
@@ -65,6 +67,54 @@ def test_map_name(self):
         self.fmt.teardown()
         self.assertFalse(self.fmt.status)
 
+    def test_add_remove_passphrase(self):
+        self.fmt.device = self.loop_devices[0]
+
+        # create the luks format
+        self.fmt.create()
+
+        # add a new passphrase
+        self.fmt.add_passphrase("password2")
+
+        # we should now be able to setup the format using this passphrase
+        self.fmt.passphrase = "password2"
+        self.fmt.setup()
+        self.fmt.teardown()
+
+        # remove the original passphrase
+        self.fmt.passphrase = "password"
+        self.fmt.remove_passphrase()
+
+        # now setup should fail
+        with self.assertRaises(LUKSError):
+            self.fmt.setup()
+
+        # setup with the new passphrase should still be possible
+        self.fmt.passphrase = "password2"
+        self.fmt.setup()
+        self.fmt.teardown()
+
+    def test_setup_keyfile(self):
+        self.fmt.device = self.loop_devices[0]
+
+        with tempfile.NamedTemporaryFile(prefix="blivet_test") as temp:
+            temp.write(b"password2")
+
+            # create the luks format with both passphrase and keyfile
+            self.fmt._key_file = temp.name
+            self.fmt.create()
+
+            # open first with just password
+            self.fmt._key_file = None
+            self.fmt.setup()
+            self.fmt.teardown()
+
+            # now with keyfile
+            self.fmt._key_file = temp.name
+            self.fmt.passphrase = None
+            self.fmt.setup()
+            self.fmt.teardown()
+
     def _luks_close(self):
         self.fmt.teardown()