tests: Add case for LUKS volume with detached header

Also, add a section to the MAINTAINERS file for detached
LUKS header, it only has a test case in it currently.

Signed-off-by: Hyman Huang <yong.huang@smartx.com>
Reviewed-by: Daniel P. Berrangé <berrange@redhat.com>
Signed-off-by: Daniel P. Berrangé <berrange@redhat.com>

MAINTAINERS

@@ -3402,6 +3402,11 @@ F: migration/dirtyrate.c
 F: migration/dirtyrate.h
 F: include/sysemu/dirtyrate.h
 
+Detached LUKS header
+M: Hyman Huang <yong.huang@smartx.com>
+S: Maintained
+F: tests/qemu-iotests/tests/luks-detached-header
+
 D-Bus
 M: Marc-André Lureau <marcandre.lureau@redhat.com>
 S: Maintained

tests/qemu-iotests/tests/luks-detached-header

@@ -0,0 +1,316 @@
+#!/usr/bin/env python3
+# group: rw auto
+#
+# Test LUKS volume with detached header
+#
+# Copyright (C) 2024 SmartX Inc.
+#
+# Authors:
+#     Hyman Huang <yong.huang@smartx.com>
+#
+# This program is free software; you can redistribute it and/or modify
+# it under the terms of the GNU General Public License as published by
+# the Free Software Foundation; either version 2 of the License, or
+# (at your option) any later version.
+#
+# This program is distributed in the hope that it will be useful,
+# but WITHOUT ANY WARRANTY; without even the implied warranty of
+# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+# GNU General Public License for more details.
+#
+# You should have received a copy of the GNU General Public License
+# along with this program.  If not, see <http://www.gnu.org/licenses/>.
+#
+
+import os
+import json
+import iotests
+from iotests import (
+    imgfmt,
+    qemu_img_create,
+    qemu_img_info,
+    QMPTestCase,
+)
+
+
+image_size = 128 * 1024 * 1024
+
+luks_img = os.path.join(iotests.test_dir, "luks.img")
+detached_header_img1 = os.path.join(iotests.test_dir, "detached_header.img1")
+detached_header_img2 = os.path.join(iotests.test_dir, "detached_header.img2")
+detached_payload_raw_img = os.path.join(
+    iotests.test_dir, "detached_payload_raw.img"
+)
+detached_payload_qcow2_img = os.path.join(
+    iotests.test_dir, "detached_payload_qcow2.img"
+)
+detached_header_raw_img = "json:" + json.dumps(
+    {
+        "driver": "luks",
+        "file": {"filename": detached_payload_raw_img},
+        "header": {
+            "filename": detached_header_img1,
+        },
+    }
+)
+detached_header_qcow2_img = "json:" + json.dumps(
+    {
+        "driver": "luks",
+        "file": {"filename": detached_payload_qcow2_img},
+        "header": {"filename": detached_header_img2},
+    }
+)
+
+secret_obj = "secret,id=sec0,data=foo"
+luks_opts = "key-secret=sec0"
+
+
+class TestDetachedLUKSHeader(QMPTestCase):
+    def setUp(self) -> None:
+        self.vm = iotests.VM()
+        self.vm.add_object(secret_obj)
+        self.vm.launch()
+
+        # 1. Create the normal LUKS disk with 128M size
+        self.vm.blockdev_create(
+            {"driver": "file", "filename": luks_img, "size": 0}
+        )
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=luks_img,
+            node_name="luks-1-storage",
+        )
+        result = self.vm.blockdev_create(
+            {
+                "driver": imgfmt,
+                "file": "luks-1-storage",
+                "key-secret": "sec0",
+                "size": image_size,
+                "iter-time": 10,
+            }
+        )
+        # None is expected
+        self.assertEqual(result, None)
+
+        # 2. Create the LUKS disk with detached header (raw)
+
+        # Create detached LUKS header
+        self.vm.blockdev_create(
+            {"driver": "file", "filename": detached_header_img1, "size": 0}
+        )
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=detached_header_img1,
+            node_name="luks-2-header-storage",
+        )
+
+        # Create detached LUKS raw payload
+        self.vm.blockdev_create(
+            {"driver": "file", "filename": detached_payload_raw_img, "size": 0}
+        )
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=detached_payload_raw_img,
+            node_name="luks-2-payload-storage",
+        )
+
+        # Format LUKS disk with detached header
+        result = self.vm.blockdev_create(
+            {
+                "driver": imgfmt,
+                "header": "luks-2-header-storage",
+                "file": "luks-2-payload-storage",
+                "key-secret": "sec0",
+                "preallocation": "full",
+                "size": image_size,
+                "iter-time": 10,
+            }
+        )
+        self.assertEqual(result, None)
+
+        self.vm.shutdown()
+
+        # 3. Create the LUKS disk with detached header (qcow2)
+
+        # Create detached LUKS header using qemu-img
+        res = qemu_img_create(
+            "-f",
+            "luks",
+            "--object",
+            secret_obj,
+            "-o",
+            luks_opts,
+            "-o",
+            "detached-header=true",
+            detached_header_img2,
+        )
+        assert res.returncode == 0
+
+        # Create detached LUKS qcow2 payload
+        res = qemu_img_create(
+            "-f", "qcow2", detached_payload_qcow2_img, str(image_size)
+        )
+        assert res.returncode == 0
+
+    def tearDown(self) -> None:
+        os.remove(luks_img)
+        os.remove(detached_header_img1)
+        os.remove(detached_header_img2)
+        os.remove(detached_payload_raw_img)
+        os.remove(detached_payload_qcow2_img)
+
+        # Check if there was any qemu-io run that failed
+        if "Pattern verification failed" in self.vm.get_log():
+            print("ERROR: Pattern verification failed:")
+            print(self.vm.get_log())
+            self.fail("qemu-io pattern verification failed")
+
+    def test_img_creation(self) -> None:
+        # Check if the images created above are expected
+
+        data = qemu_img_info(luks_img)["format-specific"]
+        self.assertEqual(data["type"], imgfmt)
+        self.assertEqual(data["data"]["detached-header"], False)
+
+        data = qemu_img_info(detached_header_raw_img)["format-specific"]
+        self.assertEqual(data["type"], imgfmt)
+        self.assertEqual(data["data"]["detached-header"], True)
+
+        data = qemu_img_info(detached_header_qcow2_img)["format-specific"]
+        self.assertEqual(data["type"], imgfmt)
+        self.assertEqual(data["data"]["detached-header"], True)
+
+        # Check if preallocation works
+        size = qemu_img_info(detached_payload_raw_img)["actual-size"]
+        self.assertGreaterEqual(size, image_size)
+
+    def test_detached_luks_header(self) -> None:
+        self.vm.launch()
+
+        # 1. Add the disk created above
+
+        # Add normal LUKS disk
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=luks_img,
+            node_name="luks-1-storage",
+        )
+        result = self.vm.qmp_log(
+            "blockdev-add",
+            driver="luks",
+            file="luks-1-storage",
+            key_secret="sec0",
+            node_name="luks-1-format",
+        )
+
+        # Expected result{ "return": {} }
+        self.assert_qmp(result, "return", {})
+
+        # Add detached LUKS header with raw payload
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=detached_header_img1,
+            node_name="luks-header1-storage",
+        )
+
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=detached_payload_raw_img,
+            node_name="luks-2-payload-raw-storage",
+        )
+
+        result = self.vm.qmp_log(
+            "blockdev-add",
+            driver=imgfmt,
+            header="luks-header1-storage",
+            file="luks-2-payload-raw-storage",
+            key_secret="sec0",
+            node_name="luks-2-payload-raw-format",
+        )
+        self.assert_qmp(result, "return", {})
+
+        # Add detached LUKS header with qcow2 payload
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=detached_header_img2,
+            node_name="luks-header2-storage",
+        )
+
+        self.vm.qmp_log(
+            "blockdev-add",
+            driver="file",
+            filename=detached_payload_qcow2_img,
+            node_name="luks-3-payload-qcow2-storage",
+        )
+
+        result = self.vm.qmp_log(
+            "blockdev-add",
+            driver=imgfmt,
+            header="luks-header2-storage",
+            file="luks-3-payload-qcow2-storage",
+            key_secret="sec0",
+            node_name="luks-3-payload-qcow2-format",
+        )
+        self.assert_qmp(result, "return", {})
+
+        # 2. Do I/O test
+
+        # Do some I/O to the image to see whether it still works
+        # (Pattern verification will be checked by tearDown())
+
+        # Normal LUKS disk
+        result = self.vm.qmp_log(
+            "human-monitor-command",
+            command_line='qemu-io luks-1-format "write -P 40 0 64k"',
+        )
+        self.assert_qmp(result, "return", "")
+
+        result = self.vm.qmp_log(
+            "human-monitor-command",
+            command_line='qemu-io luks-1-format "read -P 40 0 64k"',
+        )
+        self.assert_qmp(result, "return", "")
+
+        # Detached LUKS header with raw payload
+        cmd = 'qemu-io luks-2-payload-raw-format "write -P 41 0 64k"'
+        result = self.vm.qmp(
+            "human-monitor-command",
+            command_line=cmd
+        )
+        self.assert_qmp(result, "return", "")
+
+        cmd = 'qemu-io luks-2-payload-raw-format "read -P 41 0 64k"'
+        result = self.vm.qmp(
+            "human-monitor-command",
+            command_line=cmd
+        )
+        self.assert_qmp(result, "return", "")
+
+        # Detached LUKS header with qcow2 payload
+        cmd = 'qemu-io luks-3-payload-qcow2-format "write -P 42 0 64k"'
+        result = self.vm.qmp(
+            "human-monitor-command",
+            command_line=cmd
+        )
+        self.assert_qmp(result, "return", "")
+
+        cmd = 'qemu-io luks-3-payload-qcow2-format "read -P 42 0 64k"'
+        result = self.vm.qmp(
+            "human-monitor-command",
+            command_line=cmd
+        )
+        self.assert_qmp(result, "return", "")
+
+        self.vm.shutdown()
+
+
+if __name__ == "__main__":
+    # Test image creation and I/O
+    iotests.main(supported_fmts=["luks"], supported_protocols=["file"])

tests/qemu-iotests/tests/luks-detached-header.out

@@ -0,0 +1,5 @@
+..
+----------------------------------------------------------------------
+Ran 2 tests
+
+OK