| 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"], |
| 317 | require_hmp=True) |