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

* Use 'cryptsetup erase' to kill LUKS signature.

* Adjust unit test.

* Use own wiper for LUKS headers.

* Add comments.

* Fix tests.

* Update changelog.

* Remove 'cryptsetup erase'.

* Improve error messages.

changelogs/fragments/327-luks_device-wipe.yml

@@ -0,0 +1,2 @@
+bugfixes:
+  - "luks_device - now also runs a built-in LUKS signature cleaner on ``state=absent`` to make sure that also the secondary LUKS2 header is wiped when older versions of wipefs are used (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_NAME_REGEX = re.compile(r'\s*crypt\s+([^\s]*)\s*')
 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):
@@ -515,9 +543,17 @@ class CryptHandler(Handler):
             self.run_luks_close(name)
         result = self._run_command([wipefs_bin, '--all', device])
         if result[RETURN_CODE] != 0:
-            raise ValueError('Error while wiping luks container %s: %s'
+            raise ValueError('Error while wiping LUKS container signatures for %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 signatures for %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

@@ -58,6 +58,9 @@ def test_run_luks_remove(monkeypatch):
     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")