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