LUKS HW-OPAL support

blivet/devicelibs/crypto.py

@@ -47,6 +47,9 @@
                  "luks2": BlockDev.CryptoLUKSVersion.LUKS2}
 DEFAULT_LUKS_VERSION = "luks2"
 
+OPAL_TYPES = {"luks2-hw-opal": BlockDev.CryptoLUKSHWEncryptionType.OPAL_HW_AND_SW,
+              "luks2-hw-opal-only": BlockDev.CryptoLUKSHWEncryptionType.OPAL_HW_ONLY}
+
 DEFAULT_INTEGRITY_ALGORITHM = "crc32c"
 
 # from linux/drivers/md/dm-integrity.c

blivet/devices/storage.py

@@ -419,7 +419,7 @@ def readonly(self, value):
 
     @property
     def protected(self):
-        return self.readonly or self._protected
+        return self.readonly or self._protected or self.format.protected
 
     @protected.setter
     def protected(self, value):

blivet/formats/__init__.py

@@ -158,6 +158,7 @@ class DeviceFormat(ObjectID, metaclass=SynchronizedMeta):
     _check = False
     _hidden = False                     # hide devices with this formatting?
     _ks_mountpoint = None
+    _protected = False
 
     _resize_class = fsresize.UnimplementedFSResize
     _size_info_class = fssize.UnimplementedFSSize
@@ -685,6 +686,11 @@ def packages(self):
         """ Packages required to manage formats of this type. """
         return self._packages
 
+    @property
+    def protected(self):
+        """ Is this format protected? """
+        return self._protected
+
     @property
     def resizable(self):
         """ Can formats of this type be resized? """

blivet/formats/luks.py

@@ -100,6 +100,10 @@ def __init__(self, **kwargs):
             :type pbkdf_args: :class:`LUKS2PBKDFArgs`
             :keyword luks_sector_size: encryption sector size (use only with LUKS version 2)
             :type luks_sector_size: int
+            :keyword subsystem: LUKS subsystem
+            :type subsystem: str
+            :keyword opal_admin_passphrase: OPAL admin passphrase
+            :type opal_admin_passphrase: str
 
             .. note::
 
@@ -120,13 +124,18 @@ def __init__(self, **kwargs):
         self.map_name = kwargs.get("name")
         self.luks_version = kwargs.get("luks_version") or crypto.DEFAULT_LUKS_VERSION
 
-        if self.luks_version == "luks2":
+        self.label = kwargs.get("label") or None
+        self.subsystem = kwargs.get("subsystem") or None
+
+        self.is_opal = self.luks_version in crypto.OPAL_TYPES.keys()
+
+        if self.luks_version.startswith("luks2"):
             self._header_size = crypto.LUKS2_METADATA_SIZE
         else:
             self._header_size = crypto.LUKS1_METADATA_SIZE
         self._min_size = self._header_size
 
-        if not self.exists and self.luks_version not in crypto.LUKS_VERSIONS.keys():
+        if not self.exists and self.luks_version not in list(crypto.LUKS_VERSIONS.keys()) + list(crypto.OPAL_TYPES.keys()):
             raise ValueError("Unknown or unsupported LUKS version '%s'" % self.luks_version)
 
         if not self.exists:
@@ -176,6 +185,8 @@ def __init__(self, **kwargs):
         if self.luks_sector_size and self.luks_version != "luks2":
             raise ValueError("Sector size argument is valid only for LUKS version 2.")
 
+        self.__opal_admin_passphrase = kwargs.get("opal_admin_passphrase")
+
     def __repr__(self):
         s = DeviceFormat.__repr__(self)
         if self.__passphrase:
@@ -185,11 +196,13 @@ def __repr__(self):
         s += ("  cipher = %(cipher)s  key_size = %(key_size)s"
               "  map_name = %(map_name)s\n version = %(luks_version)s"
               "  key_file = %(key_file)s  passphrase = %(passphrase)s\n"
-              "  escrow_cert = %(escrow_cert)s  add_backup = %(backup)s" %
+              "  escrow_cert = %(escrow_cert)s  add_backup = %(backup)s\n"
+              "  label = %(label)s  subsystem = %(subsystem)s" %
               {"cipher": self.cipher, "key_size": self.key_size,
                "map_name": self.map_name, "luks_version": self.luks_version,
                "key_file": self._key_file, "passphrase": passphrase,
-               "escrow_cert": self.escrow_cert, "backup": self.add_backup_passphrase})
+               "escrow_cert": self.escrow_cert, "backup": self.add_backup_passphrase,
+               "label": self.label, "subsystem": self.subsystem})
         return s
 
     @property
@@ -216,6 +229,12 @@ def _set_passphrase(self, passphrase):
 
     passphrase = property(fset=_set_passphrase)
 
+    def _set_opal_admin_passphrase(self, opal_admin_passphrase):
+        """ Set the OPAL admin passphrase for this device. """
+        self.__opal_admin_passphrase = opal_admin_passphrase
+
+    opal_admin_passphrase = property(fset=_set_opal_admin_passphrase)
+
     @property
     def has_key(self):
         return bool((self.__passphrase not in ["", None]) or
@@ -244,6 +263,19 @@ def status(self):
             return False
         return os.path.exists("/dev/mapper/%s" % self.map_name)
 
+    @property
+    def resizable(self):
+        if self.is_opal:
+            return False
+        return super(LUKS, self).resizable
+
+    @property
+    def protected(self):
+        if self.is_opal and self.exists:
+            # cannot remove LUKS HW-OPAL without admin password
+            return True
+        return False
+
     def update_size_info(self):
         """ Update this format's current size. """
 
@@ -295,10 +327,19 @@ def _teardown(self, **kwargs):
 
         udev.settle()
 
+    # pylint: disable=unused-argument
+    def _pre_destroy(self, **kwargs):
+        if self.is_opal:
+            raise LUKSError("HW-OPAL LUKS devices cannot be destroyed.")
+        return super(LUKS, self)._pre_destroy()
+
     def _pre_resize(self):
         if self.luks_version == "luks2" and not self.has_key:
             raise LUKSError("Passphrase or key needs to be set before resizing LUKS2 format.")
 
+        if self.is_opal:
+            raise LUKSError("HW-OPAL LUKS devices cannot be resized.")
+
         super(LUKS, self)._pre_resize()
 
     def _pre_create(self, **kwargs):
@@ -312,7 +353,7 @@ def _create(self, **kwargs):
                         type=self.type, status=self.status)
         super(LUKS, self)._create(**kwargs)  # set up the event sync
 
-        if not self.pbkdf_args and self.luks_version == "luks2":
+        if not self.pbkdf_args and self.luks_version.startswith("luks2"):
             if luks_data.pbkdf_args:
                 self.pbkdf_args = luks_data.pbkdf_args
             else:
@@ -324,7 +365,7 @@ def _create(self, **kwargs):
                         luks_data.pbkdf_args = self.pbkdf_args
                         log.info("PBKDF arguments for LUKS2 not specified, using defaults with memory limit %s", mem_limit)
 
-        if not self.luks_sector_size and self.luks_version == "luks2":
+        if not self.luks_sector_size and self.luks_version.startswith("luks2"):
             self.luks_sector_size = crypto.get_optimal_luks_sector_size(self.device)
 
         if self.pbkdf_args:
@@ -348,14 +389,29 @@ def _create(self, **kwargs):
         else:
             raise LUKSError("Passphrase or key file must be set for LUKS create")
 
+        if self.is_opal:
+            if not self.__opal_admin_passphrase:
+                raise LUKSError("OPAL admin passphrase must be specified when creating LUKS HW-OPAL format")
+            opal_context = blockdev.CryptoKeyslotContext(passphrase=self.__opal_admin_passphrase)
+
         try:
-            blockdev.crypto.luks_format(self.device,
-                                        context=context,
-                                        cipher=self.cipher,
-                                        key_size=self.key_size,
-                                        min_entropy=self.min_luks_entropy,
-                                        luks_version=crypto.LUKS_VERSIONS[self.luks_version],
-                                        extra=extra)
+            if self.is_opal:
+                blockdev.crypto.opal_format(self.device,
+                                            context=context,
+                                            cipher=self.cipher if self.luks_version == "luks2-hw-opal" else None,
+                                            key_size=self.key_size if self.luks_version == "luks2-hw-opal" else 0,
+                                            min_entropy=self.min_luks_entropy,
+                                            opal_context=opal_context,
+                                            hw_encryption=crypto.OPAL_TYPES[self.luks_version],
+                                            extra=extra)
+            else:
+                blockdev.crypto.luks_format(self.device,
+                                            context=context,
+                                            cipher=self.cipher,
+                                            key_size=self.key_size,
+                                            min_entropy=self.min_luks_entropy,
+                                            luks_version=crypto.LUKS_VERSIONS[self.luks_version],
+                                            extra=extra)
         except blockdev.CryptoError as e:
             raise LUKSError(e)
 
@@ -382,6 +438,8 @@ def _post_create(self, **kwargs):
 
     @property
     def destroyable(self):
+        if self.is_opal:
+            return False
         return self._plugin.available
 
     @property

blivet/populator/helpers/luks.py

@@ -127,6 +127,18 @@ def _get_kwargs(self):
             kwargs["name"] = "luks-%s" % udev.device_get_uuid(self.data)
 
         kwargs["luks_version"] = "luks%s" % udev.device_get_format_version(self.data)
+        kwargs["label"] = udev.device_get_label(self.data)
+
+        try:
+            info = blockdev.crypto.luks_info(self.device.path)
+        except blockdev.CryptoError as e:
+            log.warning("Failed to get information about LUKS format on %s: %s", self.device, str(e))
+        else:
+            if info.hw_encryption == blockdev.CryptoLUKSHWEncryptionType.OPAL_HW_AND_SW:
+                kwargs["luks_version"] = "luks2-hw-opal"
+            elif info.hw_encryption == blockdev.CryptoLUKSHWEncryptionType.OPAL_HW_ONLY:
+                kwargs["luks_version"] = "luks2-hw-opal-only"
+
         return kwargs
 
     def run(self):

plans/blivet-gui.fmf

@@ -6,6 +6,16 @@ adjust+:
   enabled: true
 
 prepare:
+  - name: copr
+    how: shell
+    script:
+      - sudo dnf install -y 'dnf-command(copr)'
+      - sudo dnf copr enable -y @storage/blivet-daily
+      # TF prioritizes Fedora tag repo over all others, in particular our daily COPR
+      - for f in $(grep -l -r 'testing-farm-tag-repository' /etc/yum.repos.d); do sed -i '/priority/d' "$f" ;done
+      - sudo dnf -y update
+
+  - name: ansible
     how: ansible
     playbook:
         - https://raw.githubusercontent.com/storaged-project/blivet-gui/main/misc/install-test-dependencies.yml

plans/tests.fmf

@@ -10,6 +10,9 @@ prepare:
     script:
       - sudo dnf install -y 'dnf-command(copr)'
       - sudo dnf copr enable -y @storage/blivet-daily
+      # TF prioritizes Fedora tag repo over all others, in particular our daily COPR
+      - for f in $(grep -l -r 'testing-farm-tag-repository' /etc/yum.repos.d); do sed -i '/priority/d' "$f" ;done
+      - sudo dnf -y update
 
   - name: ansible
     how: ansible

tests/unit_tests/formats_tests/luks_test.py

@@ -4,6 +4,7 @@
 from blivet.formats.luks import LUKS
 from blivet.size import Size
 from blivet.static_data import luks_data
+from blivet import blockdev
 
 
 class LUKSNodevTestCase(unittest.TestCase):
@@ -128,3 +129,38 @@ def test_header_size(self):
         fmt = LUKS()
         self.assertEqual(fmt._header_size, Size("16 MiB"))
         self.assertEqual(fmt._min_size, Size("16 MiB"))
+
+    def test_luks_opal(self):
+        fmt = LUKS(exists=True)
+        self.assertFalse(fmt.is_opal)
+        self.assertFalse(fmt.protected)
+
+        fmt = LUKS(luks_version="luks2-hw-opal", exists=True)
+        self.assertTrue(fmt.is_opal)
+        self.assertTrue(fmt.protected)
+
+        fmt = LUKS(luks_version="luks2-hw-opal", passphrase="passphrase", opal_admin_passphrase="passphrase")
+        with patch("blivet.devicelibs.crypto.calculate_luks2_max_memory", return_value=None):
+            with patch("blivet.devicelibs.crypto.get_optimal_luks_sector_size", return_value=512):
+                with patch("blivet.devices.lvm.blockdev.crypto") as crypto:
+                    fmt._create()
+                    crypto.luks_format.assert_not_called()
+                    crypto.opal_format.assert_called()
+                    self.assertEqual(crypto.opal_format.call_args[1]["hw_encryption"],
+                                     blockdev.CryptoLUKSHWEncryptionType.OPAL_HW_AND_SW)
+                    self.assertEqual(crypto.opal_format.call_args[1]["cipher"], "aes-xts-plain64")
+                    self.assertEqual(crypto.opal_format.call_args[1]["key_size"], 512)
+
+        fmt = LUKS(luks_version="luks2-hw-opal-only", passphrase="passphrase", opal_admin_passphrase="passphrase")
+        with patch("blivet.devicelibs.crypto.calculate_luks2_max_memory", return_value=None):
+            with patch("blivet.devicelibs.crypto.get_optimal_luks_sector_size", return_value=512):
+                with patch("blivet.devices.lvm.blockdev.crypto") as crypto:
+                    fmt._create()
+                    crypto.luks_format.assert_not_called()
+                    crypto.opal_format.assert_called()
+                    self.assertEqual(crypto.opal_format.call_args[1]["hw_encryption"],
+                                     blockdev.CryptoLUKSHWEncryptionType.OPAL_HW_ONLY)
+
+                    # cipher and key size are not valid for HW encryption only
+                    self.assertEqual(crypto.opal_format.call_args[1]["cipher"], None)
+                    self.assertEqual(crypto.opal_format.call_args[1]["key_size"], 0)