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