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>
This commit is contained in:
Hyman Huang 2024-01-30 13:37:25 +08:00 committed by Daniel P. Berrangé
parent 0bd779e27e
commit d87b258b75
3 changed files with 326 additions and 0 deletions

View File

@ -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

View File

@ -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"])

View File

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