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

   1 #!/usr/bin/env python3
   2 # group: rw auto
   3 #
   4 # Test LUKS volume with detached header
   5 #
   6 # Copyright (C) 2024 SmartX Inc.
   7 #
   8 # Authors:
   9 #     Hyman Huang <yong.huang@smartx.com>
  10 #
  11 # This program is free software; you can redistribute it and/or modify
  12 # it under the terms of the GNU General Public License as published by
  13 # the Free Software Foundation; either version 2 of the License, or
  14 # (at your option) any later version.
  15 #
  16 # This program is distributed in the hope that it will be useful,
  17 # but WITHOUT ANY WARRANTY; without even the implied warranty of
  18 # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
  19 # GNU General Public License for more details.
  20 #
  21 # You should have received a copy of the GNU General Public License
  22 # along with this program.  If not, see <http://www.gnu.org/licenses/>.
  23 #
  24
  25 import os
  26 import json
  27 import iotests
  28 from iotests import (
  29     imgfmt,
  30     qemu_img_create,
  31     qemu_img_info,
  32     QMPTestCase,
  33 )
  34
  35
  36 image_size = 128 * 1024 * 1024
  37
  38 luks_img = os.path.join(iotests.test_dir, "luks.img")
  39 detached_header_img1 = os.path.join(iotests.test_dir, "detached_header.img1")
  40 detached_header_img2 = os.path.join(iotests.test_dir, "detached_header.img2")
  41 detached_payload_raw_img = os.path.join(
  42     iotests.test_dir, "detached_payload_raw.img"
  43 )
  44 detached_payload_qcow2_img = os.path.join(
  45     iotests.test_dir, "detached_payload_qcow2.img"
  46 )
  47 detached_header_raw_img = "json:" + json.dumps(
  48     {
  49         "driver": "luks",
  50         "file": {"filename": detached_payload_raw_img},
  51         "header": {
  52             "filename": detached_header_img1,
  53         },
  54     }
  55 )
  56 detached_header_qcow2_img = "json:" + json.dumps(
  57     {
  58         "driver": "luks",
  59         "file": {"filename": detached_payload_qcow2_img},
  60         "header": {"filename": detached_header_img2},
  61     }
  62 )
  63
  64 secret_obj = "secret,id=sec0,data=foo"
  65 luks_opts = "key-secret=sec0"
  66
  67
  68 class TestDetachedLUKSHeader(QMPTestCase):
  69     def setUp(self) -> None:
  70         self.vm = iotests.VM()
  71         self.vm.add_object(secret_obj)
  72         self.vm.launch()
  73
  74         # 1. Create the normal LUKS disk with 128M size
  75         self.vm.blockdev_create(
  76             {"driver": "file", "filename": luks_img, "size": 0}
  77         )
  78         self.vm.qmp_log(
  79             "blockdev-add",
  80             driver="file",
  81             filename=luks_img,
  82             node_name="luks-1-storage",
  83         )
  84         result = self.vm.blockdev_create(
  85             {
  86                 "driver": imgfmt,
  87                 "file": "luks-1-storage",
  88                 "key-secret": "sec0",
  89                 "size": image_size,
  90                 "iter-time": 10,
  91             }
  92         )
  93         # None is expected
  94         self.assertEqual(result, None)
  95
  96         # 2. Create the LUKS disk with detached header (raw)
  97
  98         # Create detached LUKS header
  99         self.vm.blockdev_create(
 100             {"driver": "file", "filename": detached_header_img1, "size": 0}
 101         )
 102         self.vm.qmp_log(
 103             "blockdev-add",
 104             driver="file",
 105             filename=detached_header_img1,
 106             node_name="luks-2-header-storage",
 107         )
 108
 109         # Create detached LUKS raw payload
 110         self.vm.blockdev_create(
 111             {"driver": "file", "filename": detached_payload_raw_img, "size": 0}
 112         )
 113         self.vm.qmp_log(
 114             "blockdev-add",
 115             driver="file",
 116             filename=detached_payload_raw_img,
 117             node_name="luks-2-payload-storage",
 118         )
 119
 120         # Format LUKS disk with detached header
 121         result = self.vm.blockdev_create(
 122             {
 123                 "driver": imgfmt,
 124                 "header": "luks-2-header-storage",
 125                 "file": "luks-2-payload-storage",
 126                 "key-secret": "sec0",
 127                 "preallocation": "full",
 128                 "size": image_size,
 129                 "iter-time": 10,
 130             }
 131         )
 132         self.assertEqual(result, None)
 133
 134         self.vm.shutdown()
 135
 136         # 3. Create the LUKS disk with detached header (qcow2)
 137
 138         # Create detached LUKS header using qemu-img
 139         res = qemu_img_create(
 140             "-f",
 141             "luks",
 142             "--object",
 143             secret_obj,
 144             "-o",
 145             luks_opts,
 146             "-o",
 147             "detached-header=true",
 148             detached_header_img2,
 149         )
 150         assert res.returncode == 0
 151
 152         # Create detached LUKS qcow2 payload
 153         res = qemu_img_create(
 154             "-f", "qcow2", detached_payload_qcow2_img, str(image_size)
 155         )
 156         assert res.returncode == 0
 157
 158     def tearDown(self) -> None:
 159         os.remove(luks_img)
 160         os.remove(detached_header_img1)
 161         os.remove(detached_header_img2)
 162         os.remove(detached_payload_raw_img)
 163         os.remove(detached_payload_qcow2_img)
 164
 165         # Check if there was any qemu-io run that failed
 166         if "Pattern verification failed" in self.vm.get_log():
 167             print("ERROR: Pattern verification failed:")
 168             print(self.vm.get_log())
 169             self.fail("qemu-io pattern verification failed")
 170
 171     def test_img_creation(self) -> None:
 172         # Check if the images created above are expected
 173
 174         data = qemu_img_info(luks_img)["format-specific"]
 175         self.assertEqual(data["type"], imgfmt)
 176         self.assertEqual(data["data"]["detached-header"], False)
 177
 178         data = qemu_img_info(detached_header_raw_img)["format-specific"]
 179         self.assertEqual(data["type"], imgfmt)
 180         self.assertEqual(data["data"]["detached-header"], True)
 181
 182         data = qemu_img_info(detached_header_qcow2_img)["format-specific"]
 183         self.assertEqual(data["type"], imgfmt)
 184         self.assertEqual(data["data"]["detached-header"], True)
 185
 186         # Check if preallocation works
 187         size = qemu_img_info(detached_payload_raw_img)["actual-size"]
 188         self.assertGreaterEqual(size, image_size)
 189
 190     def test_detached_luks_header(self) -> None:
 191         self.vm.launch()
 192
 193         # 1. Add the disk created above
 194
 195         # Add normal LUKS disk
 196         self.vm.qmp_log(
 197             "blockdev-add",
 198             driver="file",
 199             filename=luks_img,
 200             node_name="luks-1-storage",
 201         )
 202         result = self.vm.qmp_log(
 203             "blockdev-add",
 204             driver="luks",
 205             file="luks-1-storage",
 206             key_secret="sec0",
 207             node_name="luks-1-format",
 208         )
 209
 210         # Expected result{ "return": {} }
 211         self.assert_qmp(result, "return", {})
 212
 213         # Add detached LUKS header with raw payload
 214         self.vm.qmp_log(
 215             "blockdev-add",
 216             driver="file",
 217             filename=detached_header_img1,
 218             node_name="luks-header1-storage",
 219         )
 220
 221         self.vm.qmp_log(
 222             "blockdev-add",
 223             driver="file",
 224             filename=detached_payload_raw_img,
 225             node_name="luks-2-payload-raw-storage",
 226         )
 227
 228         result = self.vm.qmp_log(
 229             "blockdev-add",
 230             driver=imgfmt,
 231             header="luks-header1-storage",
 232             file="luks-2-payload-raw-storage",
 233             key_secret="sec0",
 234             node_name="luks-2-payload-raw-format",
 235         )
 236         self.assert_qmp(result, "return", {})
 237
 238         # Add detached LUKS header with qcow2 payload
 239         self.vm.qmp_log(
 240             "blockdev-add",
 241             driver="file",
 242             filename=detached_header_img2,
 243             node_name="luks-header2-storage",
 244         )
 245
 246         self.vm.qmp_log(
 247             "blockdev-add",
 248             driver="file",
 249             filename=detached_payload_qcow2_img,
 250             node_name="luks-3-payload-qcow2-storage",
 251         )
 252
 253         result = self.vm.qmp_log(
 254             "blockdev-add",
 255             driver=imgfmt,
 256             header="luks-header2-storage",
 257             file="luks-3-payload-qcow2-storage",
 258             key_secret="sec0",
 259             node_name="luks-3-payload-qcow2-format",
 260         )
 261         self.assert_qmp(result, "return", {})
 262
 263         # 2. Do I/O test
 264
 265         # Do some I/O to the image to see whether it still works
 266         # (Pattern verification will be checked by tearDown())
 267
 268         # Normal LUKS disk
 269         result = self.vm.qmp_log(
 270             "human-monitor-command",
 271             command_line='qemu-io luks-1-format "write -P 40 0 64k"',
 272         )
 273         self.assert_qmp(result, "return", "")
 274
 275         result = self.vm.qmp_log(
 276             "human-monitor-command",
 277             command_line='qemu-io luks-1-format "read -P 40 0 64k"',
 278         )
 279         self.assert_qmp(result, "return", "")
 280
 281         # Detached LUKS header with raw payload
 282         cmd = 'qemu-io luks-2-payload-raw-format "write -P 41 0 64k"'
 283         result = self.vm.qmp(
 284             "human-monitor-command",
 285             command_line=cmd
 286         )
 287         self.assert_qmp(result, "return", "")
 288
 289         cmd = 'qemu-io luks-2-payload-raw-format "read -P 41 0 64k"'
 290         result = self.vm.qmp(
 291             "human-monitor-command",
 292             command_line=cmd
 293         )
 294         self.assert_qmp(result, "return", "")
 295
 296         # Detached LUKS header with qcow2 payload
 297         cmd = 'qemu-io luks-3-payload-qcow2-format "write -P 42 0 64k"'
 298         result = self.vm.qmp(
 299             "human-monitor-command",
 300             command_line=cmd
 301         )
 302         self.assert_qmp(result, "return", "")
 303
 304         cmd = 'qemu-io luks-3-payload-qcow2-format "read -P 42 0 64k"'
 305         result = self.vm.qmp(
 306             "human-monitor-command",
 307             command_line=cmd
 308         )
 309         self.assert_qmp(result, "return", "")
 310
 311         self.vm.shutdown()
 312
 313
 314 if __name__ == "__main__":
 315     # Test image creation and I/O
 316     iotests.main(supported_fmts=["luks"], supported_protocols=["file"])