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 25import os 26import json 27import iotests 28from iotests import ( 29 imgfmt, 30 qemu_img_create, 31 qemu_img_info, 32 QMPTestCase, 33) 34 35 36image_size = 128 * 1024 * 1024 37 38luks_img = os.path.join(iotests.test_dir, "luks.img") 39detached_header_img1 = os.path.join(iotests.test_dir, "detached_header.img1") 40detached_header_img2 = os.path.join(iotests.test_dir, "detached_header.img2") 41detached_payload_raw_img = os.path.join( 42 iotests.test_dir, "detached_payload_raw.img" 43) 44detached_payload_qcow2_img = os.path.join( 45 iotests.test_dir, "detached_payload_qcow2.img" 46) 47detached_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) 56detached_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 64secret_obj = "secret,id=sec0,data=foo" 65luks_opts = "key-secret=sec0" 66 67 68class 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 314if __name__ == "__main__": 315 # Test image creation and I/O 316 iotests.main(supported_fmts=["luks"], supported_protocols=["file"]) 317