1#!/usr/bin/env python3 2# group: rw sudo 3# 4# Copyright (C) 2016 Red Hat, Inc. 5# 6# This program is free software; you can redistribute it and/or modify 7# it under the terms of the GNU General Public License as published by 8# the Free Software Foundation; either version 2 of the License, or 9# (at your option) any later version. 10# 11# This program is distributed in the hope that it will be useful, 12# but WITHOUT ANY WARRANTY; without even the implied warranty of 13# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the 14# GNU General Public License for more details. 15# 16# You should have received a copy of the GNU General Public License 17# along with this program. If not, see <http://www.gnu.org/licenses/>. 18# 19# Creator/Owner: Daniel P. Berrange <berrange@redhat.com> 20# 21# Exercise the QEMU 'luks' block driver to validate interoperability 22# with the Linux dm-crypt + cryptsetup implementation 23 24import subprocess 25import os 26import os.path 27 28import base64 29 30import iotests 31 32 33class LUKSConfig(object): 34 """Represent configuration parameters for a single LUKS 35 setup to be tested""" 36 37 def __init__(self, name, cipher, keylen, mode, ivgen, 38 ivgen_hash, hash, password=None, passwords=None): 39 40 self.name = name 41 self.cipher = cipher 42 self.keylen = keylen 43 self.mode = mode 44 self.ivgen = ivgen 45 self.ivgen_hash = ivgen_hash 46 self.hash = hash 47 48 if passwords is not None: 49 self.passwords = passwords 50 else: 51 self.passwords = {} 52 53 if password is None: 54 self.passwords["0"] = "123456" 55 else: 56 self.passwords["0"] = password 57 58 def __repr__(self): 59 return self.name 60 61 def image_name(self): 62 return "luks-%s.img" % self.name 63 64 def image_path(self): 65 return os.path.join(iotests.test_dir, self.image_name()) 66 67 def device_name(self): 68 return "qiotest-145-%s" % self.name 69 70 def device_path(self): 71 return "/dev/mapper/" + self.device_name() 72 73 def first_password(self): 74 for i in range(8): 75 slot = str(i) 76 if slot in self.passwords: 77 return (self.passwords[slot], slot) 78 raise Exception("No password found") 79 80 def first_password_base64(self): 81 (pw, slot) = self.first_password() 82 return base64.b64encode(pw.encode('ascii')).decode('ascii') 83 84 def active_slots(self): 85 slots = [] 86 for i in range(8): 87 slot = str(i) 88 if slot in self.passwords: 89 slots.append(slot) 90 return slots 91 92def verify_passwordless_sudo(): 93 """Check whether sudo is configured to allow 94 password-less access to commands""" 95 96 args = ["sudo", "-n", "/bin/true"] 97 98 proc = subprocess.Popen(args, 99 stdin=subprocess.PIPE, 100 stdout=subprocess.PIPE, 101 stderr=subprocess.STDOUT, 102 universal_newlines=True) 103 104 msg = proc.communicate()[0] 105 106 if proc.returncode != 0: 107 iotests.notrun('requires password-less sudo access: %s' % msg) 108 109 110def cryptsetup(args, password=None): 111 """Run the cryptsetup command in batch mode""" 112 113 fullargs = ["sudo", "cryptsetup", "-q", "-v"] 114 fullargs.extend(args) 115 116 iotests.log(" ".join(fullargs), filters=[iotests.filter_test_dir]) 117 proc = subprocess.Popen(fullargs, 118 stdin=subprocess.PIPE, 119 stdout=subprocess.PIPE, 120 stderr=subprocess.STDOUT, 121 universal_newlines=True) 122 123 msg = proc.communicate(password)[0] 124 125 if proc.returncode != 0: 126 raise Exception(msg) 127 128 129def cryptsetup_add_password(config, slot): 130 """Add another password to a LUKS key slot""" 131 132 (password, mainslot) = config.first_password() 133 134 pwfile = os.path.join(iotests.test_dir, "passwd.txt") 135 with open(pwfile, "w") as fh: 136 fh.write(config.passwords[slot]) 137 138 try: 139 args = ["luksAddKey", config.image_path(), 140 "--key-slot", slot, 141 "--key-file", "-", 142 "--iter-time", "10", 143 pwfile] 144 145 cryptsetup(args, password) 146 finally: 147 os.unlink(pwfile) 148 149 150def cryptsetup_format(config): 151 """Format a new LUKS volume with cryptsetup, adding the 152 first key slot only""" 153 154 (password, slot) = config.first_password() 155 156 args = ["luksFormat", "--type", "luks1"] 157 cipher = config.cipher + "-" + config.mode + "-" + config.ivgen 158 if config.ivgen_hash is not None: 159 cipher = cipher + ":" + config.ivgen_hash 160 elif config.ivgen == "essiv": 161 cipher = cipher + ":" + "sha256" 162 args.extend(["--cipher", cipher]) 163 if config.mode == "xts": 164 args.extend(["--key-size", str(config.keylen * 2)]) 165 else: 166 args.extend(["--key-size", str(config.keylen)]) 167 if config.hash is not None: 168 args.extend(["--hash", config.hash]) 169 args.extend(["--key-slot", slot]) 170 args.extend(["--key-file", "-"]) 171 args.extend(["--iter-time", "10"]) 172 args.append(config.image_path()) 173 174 cryptsetup(args, password) 175 176 177def chown(config): 178 """Set the ownership of a open LUKS device to this user""" 179 180 path = config.device_path() 181 182 args = ["sudo", "chown", "%d:%d" % (os.getuid(), os.getgid()), path] 183 iotests.log(" ".join(args), filters=[iotests.filter_chown]) 184 proc = subprocess.Popen(args, 185 stdin=subprocess.PIPE, 186 stdout=subprocess.PIPE, 187 stderr=subprocess.STDOUT) 188 189 msg = proc.communicate()[0] 190 191 if proc.returncode != 0: 192 raise Exception(msg) 193 194 195def cryptsetup_open(config): 196 """Open an image as a LUKS device""" 197 198 (password, slot) = config.first_password() 199 200 args = ["luksOpen", config.image_path(), config.device_name()] 201 202 cryptsetup(args, password) 203 204 205def cryptsetup_close(config): 206 """Close an active LUKS device """ 207 208 args = ["luksClose", config.device_name()] 209 cryptsetup(args) 210 211 212def delete_image(config): 213 """Delete a disk image""" 214 215 try: 216 os.unlink(config.image_path()) 217 iotests.log("unlink %s" % config.image_path(), 218 filters=[iotests.filter_test_dir]) 219 except Exception as e: 220 pass 221 222 223def create_image(config, size_mb): 224 """Create a bare disk image with requested size""" 225 226 delete_image(config) 227 iotests.log("truncate %s --size %dMB" % (config.image_path(), size_mb), 228 filters=[iotests.filter_test_dir]) 229 with open(config.image_path(), "w") as fn: 230 fn.truncate(size_mb * 1024 * 1024) 231 232 233def check_cipher_support(config, output): 234 """Check the output of qemu-img or qemu-io for mention of the respective 235 cipher algorithm being unsupported, and if so, skip this test. 236 (Returns `output` for convenience.)""" 237 238 if 'Unsupported cipher algorithm' in output: 239 iotests.notrun('Unsupported cipher algorithm ' 240 f'{config.cipher}-{config.keylen}-{config.mode}; ' 241 'consider configuring qemu with a different crypto ' 242 'backend') 243 return output 244 245def qemu_img_create(config, size_mb): 246 """Create and format a disk image with LUKS using qemu-img""" 247 248 opts = [ 249 "key-secret=sec0", 250 "iter-time=10", 251 "cipher-alg=%s-%d" % (config.cipher, config.keylen), 252 "cipher-mode=%s" % config.mode, 253 "ivgen-alg=%s" % config.ivgen, 254 "hash-alg=%s" % config.hash, 255 ] 256 if config.ivgen_hash is not None: 257 opts.append("ivgen-hash-alg=%s" % config.ivgen_hash) 258 259 args = ["create", "-f", "luks", 260 "--object", 261 ("secret,id=sec0,data=%s,format=base64" % 262 config.first_password_base64()), 263 "-o", ",".join(opts), 264 config.image_path(), 265 "%dM" % size_mb] 266 267 iotests.log("qemu-img " + " ".join(args), filters=[iotests.filter_test_dir]) 268 try: 269 iotests.qemu_img(*args) 270 except subprocess.CalledProcessError as exc: 271 check_cipher_support(config, exc.output) 272 raise 273 274def qemu_io_image_args(config, dev=False): 275 """Get the args for access an image or device with qemu-io""" 276 277 if dev: 278 return [ 279 "--image-opts", 280 "driver=host_device,filename=%s" % config.device_path()] 281 else: 282 return [ 283 "--object", 284 ("secret,id=sec0,data=%s,format=base64" % 285 config.first_password_base64()), 286 "--image-opts", 287 ("driver=luks,key-secret=sec0,file.filename=%s" % 288 config.image_path())] 289 290def qemu_io_write_pattern(config, pattern, offset_mb, size_mb, dev=False): 291 """Write a pattern of data to a LUKS image or device""" 292 293 if dev: 294 chown(config) 295 args = ["-c", "write -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)] 296 args.extend(qemu_io_image_args(config, dev)) 297 iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir]) 298 output = iotests.qemu_io(*args, check=False).stdout 299 iotests.log(check_cipher_support(config, output), 300 filters=[iotests.filter_test_dir, iotests.filter_qemu_io]) 301 302 303def qemu_io_read_pattern(config, pattern, offset_mb, size_mb, dev=False): 304 """Read a pattern of data to a LUKS image or device""" 305 306 if dev: 307 chown(config) 308 args = ["-c", "read -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)] 309 args.extend(qemu_io_image_args(config, dev)) 310 iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir]) 311 output = iotests.qemu_io(*args, check=False).stdout 312 iotests.log(check_cipher_support(config, output), 313 filters=[iotests.filter_test_dir, iotests.filter_qemu_io]) 314 315 316def test_once(config, qemu_img=False): 317 """Run the test with a desired LUKS configuration. Can either 318 use qemu-img for creating the initial volume, or cryptsetup, 319 in order to test interoperability in both directions""" 320 321 iotests.log("# ================= %s %s =================" % ( 322 "qemu-img" if qemu_img else "dm-crypt", config)) 323 324 oneKB = 1024 325 oneMB = oneKB * 1024 326 oneGB = oneMB * 1024 327 oneTB = oneGB * 1024 328 329 # 4 TB, so that we pass the 32-bit sector number boundary. 330 # Important for testing correctness of some IV generators 331 # The files are sparse, so not actually using this much space 332 image_size = 4 * oneTB 333 if qemu_img: 334 iotests.log("# Create image") 335 qemu_img_create(config, image_size // oneMB) 336 else: 337 iotests.log("# Create image") 338 create_image(config, image_size // oneMB) 339 340 lowOffsetMB = 100 341 highOffsetMB = 3 * oneTB // oneMB 342 343 try: 344 if not qemu_img: 345 iotests.log("# Format image") 346 cryptsetup_format(config) 347 348 for slot in config.active_slots()[1:]: 349 iotests.log("# Add password slot %s" % slot) 350 cryptsetup_add_password(config, slot) 351 352 # First we'll open the image using cryptsetup and write a 353 # known pattern of data that we'll then verify with QEMU 354 355 iotests.log("# Open dev") 356 cryptsetup_open(config) 357 358 try: 359 iotests.log("# Write test pattern 0xa7") 360 qemu_io_write_pattern(config, 0xa7, lowOffsetMB, 10, dev=True) 361 iotests.log("# Write test pattern 0x13") 362 qemu_io_write_pattern(config, 0x13, highOffsetMB, 10, dev=True) 363 finally: 364 iotests.log("# Close dev") 365 cryptsetup_close(config) 366 367 # Ok, now we're using QEMU to verify the pattern just 368 # written via dm-crypt 369 370 iotests.log("# Read test pattern 0xa7") 371 qemu_io_read_pattern(config, 0xa7, lowOffsetMB, 10, dev=False) 372 iotests.log("# Read test pattern 0x13") 373 qemu_io_read_pattern(config, 0x13, highOffsetMB, 10, dev=False) 374 375 376 # Write a new pattern to the image, which we'll later 377 # verify with dm-crypt 378 iotests.log("# Write test pattern 0x91") 379 qemu_io_write_pattern(config, 0x91, lowOffsetMB, 10, dev=False) 380 iotests.log("# Write test pattern 0x5e") 381 qemu_io_write_pattern(config, 0x5e, highOffsetMB, 10, dev=False) 382 383 384 # Now we're opening the image with dm-crypt once more 385 # and verifying what QEMU wrote, completing the circle 386 iotests.log("# Open dev") 387 cryptsetup_open(config) 388 389 try: 390 iotests.log("# Read test pattern 0x91") 391 qemu_io_read_pattern(config, 0x91, lowOffsetMB, 10, dev=True) 392 iotests.log("# Read test pattern 0x5e") 393 qemu_io_read_pattern(config, 0x5e, highOffsetMB, 10, dev=True) 394 finally: 395 iotests.log("# Close dev") 396 cryptsetup_close(config) 397 finally: 398 iotests.log("# Delete image") 399 delete_image(config) 400 print() 401 402 403# Obviously we only work with the luks image format 404iotests.script_initialize(supported_fmts=['luks']) 405 406# We need sudo in order to run cryptsetup to create 407# dm-crypt devices. This is safe to use on any 408# machine, since all dm-crypt devices are backed 409# by newly created plain files, and have a dm-crypt 410# name prefix of 'qiotest' to avoid clashing with 411# user LUKS volumes 412verify_passwordless_sudo() 413 414 415# If we look at all permutations of cipher, key size, 416# mode, ivgen, hash, there are ~1000 possible configs. 417# 418# We certainly don't want/need to test every permutation 419# to get good validation of interoperability between QEMU 420# and dm-crypt/cryptsetup. 421# 422# The configs below are a representative set that aim to 423# exercise each axis of configurability. 424# 425configs = [ 426 # A common LUKS default 427 LUKSConfig("aes-256-xts-plain64-sha1", 428 "aes", 256, "xts", "plain64", None, "sha1"), 429 430 431 # LUKS default but diff ciphers 432 LUKSConfig("twofish-256-xts-plain64-sha1", 433 "twofish", 256, "xts", "plain64", None, "sha1"), 434 LUKSConfig("serpent-256-xts-plain64-sha1", 435 "serpent", 256, "xts", "plain64", None, "sha1"), 436 # Should really be xts, but kernel doesn't support xts+cast5 437 # nor does it do essiv+cast5 438 LUKSConfig("cast5-128-cbc-plain64-sha1", 439 "cast5", 128, "cbc", "plain64", None, "sha1"), 440 LUKSConfig("cast6-256-xts-plain64-sha1", 441 "cast6", 256, "xts", "plain64", None, "sha1"), 442 443 444 # LUKS default but diff modes / ivgens 445 LUKSConfig("aes-256-cbc-plain-sha1", 446 "aes", 256, "cbc", "plain", None, "sha1"), 447 LUKSConfig("aes-256-cbc-plain64-sha1", 448 "aes", 256, "cbc", "plain64", None, "sha1"), 449 LUKSConfig("aes-256-cbc-essiv-sha256-sha1", 450 "aes", 256, "cbc", "essiv", "sha256", "sha1"), 451 LUKSConfig("aes-256-xts-essiv-sha256-sha1", 452 "aes", 256, "xts", "essiv", "sha256", "sha1"), 453 454 455 # LUKS default but smaller key sizes 456 LUKSConfig("aes-128-xts-plain64-sha256-sha1", 457 "aes", 128, "xts", "plain64", None, "sha1"), 458 LUKSConfig("aes-192-xts-plain64-sha256-sha1", 459 "aes", 192, "xts", "plain64", None, "sha1"), 460 461 LUKSConfig("twofish-128-xts-plain64-sha1", 462 "twofish", 128, "xts", "plain64", None, "sha1"), 463 LUKSConfig("twofish-192-xts-plain64-sha1", 464 "twofish", 192, "xts", "plain64", None, "sha1"), 465 466 LUKSConfig("serpent-128-xts-plain64-sha1", 467 "serpent", 128, "xts", "plain64", None, "sha1"), 468 LUKSConfig("serpent-192-xts-plain64-sha1", 469 "serpent", 192, "xts", "plain64", None, "sha1"), 470 471 LUKSConfig("cast6-128-xts-plain64-sha1", 472 "cast6", 128, "xts", "plain", None, "sha1"), 473 LUKSConfig("cast6-192-xts-plain64-sha1", 474 "cast6", 192, "xts", "plain64", None, "sha1"), 475 476 477 # LUKS default but diff hash 478 LUKSConfig("aes-256-xts-plain64-sha224", 479 "aes", 256, "xts", "plain64", None, "sha224"), 480 LUKSConfig("aes-256-xts-plain64-sha256", 481 "aes", 256, "xts", "plain64", None, "sha256"), 482 LUKSConfig("aes-256-xts-plain64-sha384", 483 "aes", 256, "xts", "plain64", None, "sha384"), 484 LUKSConfig("aes-256-xts-plain64-sha512", 485 "aes", 256, "xts", "plain64", None, "sha512"), 486 LUKSConfig("aes-256-xts-plain64-ripemd160", 487 "aes", 256, "xts", "plain64", None, "ripemd160"), 488 489 # Password in slot 3 490 LUKSConfig("aes-256-xts-plain-sha1-pwslot3", 491 "aes", 256, "xts", "plain", None, "sha1", 492 passwords={ 493 "3": "slot3", 494 }), 495 496 # Passwords in every slot 497 LUKSConfig("aes-256-xts-plain-sha1-pwallslots", 498 "aes", 256, "xts", "plain", None, "sha1", 499 passwords={ 500 "0": "slot1", 501 "1": "slot1", 502 "2": "slot2", 503 "3": "slot3", 504 "4": "slot4", 505 "5": "slot5", 506 "6": "slot6", 507 "7": "slot7", 508 }), 509 510 # Check handling of default hash alg (sha256) with essiv 511 LUKSConfig("aes-256-cbc-essiv-auto-sha1", 512 "aes", 256, "cbc", "essiv", None, "sha1"), 513 514 # Check that a useless hash provided for 'plain64' iv gen 515 # is ignored and no error raised 516 LUKSConfig("aes-256-cbc-plain64-sha256-sha1", 517 "aes", 256, "cbc", "plain64", "sha256", "sha1"), 518 519] 520 521unsupported_configs = [ 522 # We don't have a cast-6 cipher impl for QEMU yet 523 "cast6-256-xts-plain64-sha1", 524 "cast6-128-xts-plain64-sha1", 525 "cast6-192-xts-plain64-sha1", 526 527 # GCrypt doesn't support Twofish with 192 bit key 528 "twofish-192-xts-plain64-sha1", 529] 530 531# Optionally test only the configurations in the LUKS_CONFIG 532# environment variable 533tested_configs = None 534if "LUKS_CONFIG" in os.environ: 535 tested_configs = os.environ["LUKS_CONFIG"].split(",") 536 537for config in configs: 538 if config.name in unsupported_configs: 539 iotests.log("Skipping %s (config not supported)" % config.name) 540 continue 541 542 if tested_configs is not None and config.name not in tested_configs: 543 iotests.log("Skipping %s (by user request)" % config.name) 544 continue 545 546 test_once(config, qemu_img=False) 547 548 # XXX we should support setting passwords in a non-0 549 # key slot with 'qemu-img create' in future 550 (pw, slot) = config.first_password() 551 if slot == "0": 552 test_once(config, qemu_img=True) 553