luks_device: add built-in signature wiper to work around older wipefs versions with LUKS2 containers

changelogs/fragments/327-luks_device-wipe.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - "luks_device - on ``state=absent`` now also runs ``cryptsetup erase`` as well as a built-in LUKS signature cleaner to make sure the device is no longer accepted as a LUKS device (https://github.com/ansible-collections/community.crypto/issues/326, https://github.com/ansible-collections/community.crypto/pull/327)."

plugins/modules/luks_device.py

@@ -356,6 +356,34 @@
 LUKS_DEVICE_REGEX = re.compile(r'\s*device:\s+([^\s]*)\s*')
 
 
+# See https://gitlab.com/cryptsetup/cryptsetup/-/wikis/LUKS-standard/on-disk-format.pdf
+LUKS_HEADER = b'LUKS\xba\xbe'
+LUKS_HEADER_L = 6
+# See https://gitlab.com/cryptsetup/LUKS2-docs/-/blob/master/luks2_doc_wip.pdf
+LUKS2_HEADER_OFFSETS = [0x4000, 0x8000, 0x10000, 0x20000, 0x40000, 0x80000, 0x100000, 0x200000, 0x400000]
+LUKS2_HEADER2 = b'SKUL\xba\xbe'
+
+
+def wipe_luks_headers(device):
+    wipe_offsets = []
+    with open(device, 'rb') as f:
+        # f.seek(0)
+        data = f.read(LUKS_HEADER_L)
+        if data == LUKS_HEADER:
+            wipe_offsets.append(0)
+        for offset in LUKS2_HEADER_OFFSETS:
+            f.seek(offset)
+            data = f.read(LUKS_HEADER_L)
+            if data == LUKS2_HEADER2:
+                wipe_offsets.append(offset)
+
+    if wipe_offsets:
+        with open(device, 'wb') as f:
+            for offset in wipe_offsets:
+                f.seek(offset)
+                f.write(b'\x00\x00\x00\x00\x00\x00')
+
+
 class Handler(object):
 
     def __init__(self, module):
@@ -507,17 +535,31 @@ def run_luks_close(self, name):
         if result[RETURN_CODE] != 0:
             raise ValueError('Error while closing LUKS container %s' % (name))
 
+    def run_luks_erase(self, device, ignore_errors=False):
+        result = self._run_command([self._cryptsetup_bin, 'erase', '-q', device])
+        if result[RETURN_CODE] != 0 and not ignore_errors:
+            raise ValueError('Error while erasing LUKS container %s' % (device))
+
     def run_luks_remove(self, device):
         wipefs_bin = self._module.get_bin_path('wipefs', True)
 
         name = self.get_container_name_by_device(device)
         if name is not None:
             self.run_luks_close(name)
+        self.run_luks_erase(device, ignore_errors=True)
         result = self._run_command([wipefs_bin, '--all', device])
         if result[RETURN_CODE] != 0:
             raise ValueError('Error while wiping luks container %s: %s'
                              % (device, result[STDERR]))
 
+        # For LUKS2, sometimes both `cryptsetup erase` and `wipefs` do **not**
+        # erase all LUKS signatures (they seem to miss the second header). That's
+        # why we do it ourselves here.
+        try:
+            wipe_luks_headers(device)
+        except Exception as exc:
+            raise ValueError('Error while wiping luks container %s: %s' % (device, exc))
+
     def run_luks_add_key(self, device, keyfile, passphrase, new_keyfile,
                          new_passphrase, pbkdf):
         ''' Add new key from a keyfile or passphrase to given 'device';

tests/unit/plugins/modules/test_luks_device.py

@@ -46,9 +46,16 @@ def test_get_container_device_by_name(monkeypatch):
 
 
 def test_run_luks_remove(monkeypatch):
+    counter = [0]
+
     def run_command_check(self, command):
-        # check that wipefs command is actually called
-        assert command[0] == "wipefs"
+        # check that 'cryptsetup erase' and 'wipefs' commands are actually called
+        if counter[0] == 0:
+            assert command[0] == "cryptsetup"
+            assert command[1] == "erase"
+        if counter[0] == 1:
+            assert command[0] == "wipefs"
+        counter[0] += 1
         return [0, "", ""]
 
     module = DummyModule()
@@ -58,6 +65,9 @@ def run_command_check(self, command):
     monkeypatch.setattr(luks_device.Handler,
                         "_run_command",
                         run_command_check)
+    monkeypatch.setattr(luks_device,
+                        "wipe_luks_headers",
+                        lambda device: True)
     crypt = luks_device.CryptHandler(module)
     crypt.run_luks_remove("dummy")