tests/qemu-iotests/149

   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 <[email protected]>
  19 #
  20 # Exercise the QEMU 'luks' block driver to validate interoperability
  21 # with the Linux dm-crypt + cryptsetup implementation
  22
  23 import subprocess
  24 import os
  25 import os.path
  26
  27 import base64
  28
  29 import iotests
  30
  31
  32 class 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
  91 def 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
 108 def 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
 126 def 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
 147 def 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
 174 def 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(msg)
 190
 191
 192 def 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
 202 def cryptsetup_close(config):
 203     """Close an active LUKS device """
 204
 205     args = ["luksClose", config.device_name()]
 206     cryptsetup(args)
 207
 208
 209 def 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
 220 def 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
 230 def 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
 255 def 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
 271 def 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     if dev:
 275         chown(config)
 276     args = ["-c", "write -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)]
 277     args.extend(qemu_io_image_args(config, dev))
 278     iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir])
 279     iotests.log(iotests.qemu_io(*args), filters=[iotests.filter_test_dir,
 280                                                  iotests.filter_qemu_io])
 281
 282
 283 def qemu_io_read_pattern(config, pattern, offset_mb, size_mb, dev=False):
 284     """Read a pattern of data to a LUKS image or device"""
 285
 286     if dev:
 287         chown(config)
 288     args = ["-c", "read -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)]
 289     args.extend(qemu_io_image_args(config, dev))
 290     iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir])
 291     iotests.log(iotests.qemu_io(*args), filters=[iotests.filter_test_dir,
 292                                                  iotests.filter_qemu_io])
 293
 294
 295 def test_once(config, qemu_img=False):
 296     """Run the test with a desired LUKS configuration. Can either
 297        use qemu-img for creating the initial volume, or cryptsetup,
 298        in order to test interoperability in both directions"""
 299
 300     iotests.log("# ================= %s %s =================" % (
 301         "qemu-img" if qemu_img else "dm-crypt", config))
 302
 303     oneKB = 1024
 304     oneMB = oneKB * 1024
 305     oneGB = oneMB * 1024
 306     oneTB = oneGB * 1024
 307
 308     # 4 TB, so that we pass the 32-bit sector number boundary.
 309     # Important for testing correctness of some IV generators
 310     # The files are sparse, so not actually using this much space
 311     image_size = 4 * oneTB
 312     if qemu_img:
 313         iotests.log("# Create image")
 314         qemu_img_create(config, image_size / oneMB)
 315     else:
 316         iotests.log("# Create image")
 317         create_image(config, image_size / oneMB)
 318
 319     lowOffsetMB = 100
 320     highOffsetMB = 3 * oneTB / oneMB
 321
 322     try:
 323         if not qemu_img:
 324             iotests.log("# Format image")
 325             cryptsetup_format(config)
 326
 327             for slot in config.active_slots()[1:]:
 328                 iotests.log("# Add password slot %s" % slot)
 329                 cryptsetup_add_password(config, slot)
 330
 331         # First we'll open the image using cryptsetup and write a
 332         # known pattern of data that we'll then verify with QEMU
 333
 334         iotests.log("# Open dev")
 335         cryptsetup_open(config)
 336
 337         try:
 338             iotests.log("# Write test pattern 0xa7")
 339             qemu_io_write_pattern(config, 0xa7, lowOffsetMB, 10, dev=True)
 340             iotests.log("# Write test pattern 0x13")
 341             qemu_io_write_pattern(config, 0x13, highOffsetMB, 10, dev=True)
 342         finally:
 343             iotests.log("# Close dev")
 344             cryptsetup_close(config)
 345
 346         # Ok, now we're using QEMU to verify the pattern just
 347         # written via dm-crypt
 348
 349         iotests.log("# Read test pattern 0xa7")
 350         qemu_io_read_pattern(config, 0xa7, lowOffsetMB, 10, dev=False)
 351         iotests.log("# Read test pattern 0x13")
 352         qemu_io_read_pattern(config, 0x13, highOffsetMB, 10, dev=False)
 353
 354
 355         # Write a new pattern to the image, which we'll later
 356         # verify with dm-crypt
 357         iotests.log("# Write test pattern 0x91")
 358         qemu_io_write_pattern(config, 0x91, lowOffsetMB, 10, dev=False)
 359         iotests.log("# Write test pattern 0x5e")
 360         qemu_io_write_pattern(config, 0x5e, highOffsetMB, 10, dev=False)
 361
 362
 363         # Now we're opening the image with dm-crypt once more
 364         # and verifying what QEMU wrote, completing the circle
 365         iotests.log("# Open dev")
 366         cryptsetup_open(config)
 367
 368         try:
 369             iotests.log("# Read test pattern 0x91")
 370             qemu_io_read_pattern(config, 0x91, lowOffsetMB, 10, dev=True)
 371             iotests.log("# Read test pattern 0x5e")
 372             qemu_io_read_pattern(config, 0x5e, highOffsetMB, 10, dev=True)
 373         finally:
 374             iotests.log("# Close dev")
 375             cryptsetup_close(config)
 376     finally:
 377         iotests.log("# Delete image")
 378         delete_image(config)
 379         print
 380
 381
 382 # Obviously we only work with the luks image format
 383 iotests.verify_image_format(supported_fmts=['luks'])
 384 iotests.verify_platform()
 385
 386 # We need sudo in order to run cryptsetup to create
 387 # dm-crypt devices. This is safe to use on any
 388 # machine, since all dm-crypt devices are backed
 389 # by newly created plain files, and have a dm-crypt
 390 # name prefix of 'qiotest' to avoid clashing with
 391 # user LUKS volumes
 392 verify_passwordless_sudo()
 393
 394
 395 # If we look at all permutations of cipher, key size,
 396 # mode, ivgen, hash, there are ~1000 possible configs.
 397 #
 398 # We certainly don't want/need to test every permutation
 399 # to get good validation of interoperability between QEMU
 400 # and dm-crypt/cryptsetup.
 401 #
 402 # The configs below are a representative set that aim to
 403 # exercise each axis of configurability.
 404 #
 405 configs = [
 406     # A common LUKS default
 407     LUKSConfig("aes-256-xts-plain64-sha1",
 408                "aes", 256, "xts", "plain64", None, "sha1"),
 409
 410
 411     # LUKS default but diff ciphers
 412     LUKSConfig("twofish-256-xts-plain64-sha1",
 413                "twofish", 256, "xts", "plain64", None, "sha1"),
 414     LUKSConfig("serpent-256-xts-plain64-sha1",
 415                "serpent", 256, "xts", "plain64", None, "sha1"),
 416     # Should really be xts, but kernel doesn't support xts+cast5
 417     # nor does it do essiv+cast5
 418     LUKSConfig("cast5-128-cbc-plain64-sha1",
 419                "cast5", 128, "cbc", "plain64", None, "sha1"),
 420     LUKSConfig("cast6-256-xts-plain64-sha1",
 421                "cast6", 256, "xts", "plain64", None, "sha1"),
 422
 423
 424     # LUKS default but diff modes / ivgens
 425     LUKSConfig("aes-256-cbc-plain-sha1",
 426                "aes", 256, "cbc", "plain", None, "sha1"),
 427     LUKSConfig("aes-256-cbc-plain64-sha1",
 428                "aes", 256, "cbc", "plain64", None, "sha1"),
 429     LUKSConfig("aes-256-cbc-essiv-sha256-sha1",
 430                "aes", 256, "cbc", "essiv", "sha256", "sha1"),
 431     LUKSConfig("aes-256-xts-essiv-sha256-sha1",
 432                "aes", 256, "xts", "essiv", "sha256", "sha1"),
 433
 434
 435     # LUKS default but smaller key sizes
 436     LUKSConfig("aes-128-xts-plain64-sha256-sha1",
 437                "aes", 128, "xts", "plain64", None, "sha1"),
 438     LUKSConfig("aes-192-xts-plain64-sha256-sha1",
 439                "aes", 192, "xts", "plain64", None, "sha1"),
 440
 441     LUKSConfig("twofish-128-xts-plain64-sha1",
 442                "twofish", 128, "xts", "plain64", None, "sha1"),
 443     LUKSConfig("twofish-192-xts-plain64-sha1",
 444                "twofish", 192, "xts", "plain64", None, "sha1"),
 445
 446     LUKSConfig("serpent-128-xts-plain64-sha1",
 447                "serpent", 128, "xts", "plain64", None, "sha1"),
 448     LUKSConfig("serpent-192-xts-plain64-sha1",
 449                "serpent", 192, "xts", "plain64", None, "sha1"),
 450
 451     LUKSConfig("cast6-128-xts-plain64-sha1",
 452                "cast6", 128, "xts", "plain", None, "sha1"),
 453     LUKSConfig("cast6-192-xts-plain64-sha1",
 454                "cast6", 192, "xts", "plain64", None, "sha1"),
 455
 456
 457     # LUKS default but diff hash
 458     LUKSConfig("aes-256-xts-plain64-sha224",
 459                "aes", 256, "xts", "plain64", None, "sha224"),
 460     LUKSConfig("aes-256-xts-plain64-sha256",
 461                "aes", 256, "xts", "plain64", None, "sha256"),
 462     LUKSConfig("aes-256-xts-plain64-sha384",
 463                "aes", 256, "xts", "plain64", None, "sha384"),
 464     LUKSConfig("aes-256-xts-plain64-sha512",
 465                "aes", 256, "xts", "plain64", None, "sha512"),
 466     LUKSConfig("aes-256-xts-plain64-ripemd160",
 467                "aes", 256, "xts", "plain64", None, "ripemd160"),
 468
 469     # Password in slot 3
 470     LUKSConfig("aes-256-xts-plain-sha1-pwslot3",
 471                "aes", 256, "xts", "plain", None, "sha1",
 472                passwords={
 473                    "3": "slot3",
 474                }),
 475
 476     # Passwords in every slot
 477     LUKSConfig("aes-256-xts-plain-sha1-pwallslots",
 478                "aes", 256, "xts", "plain", None, "sha1",
 479                passwords={
 480                    "0": "slot1",
 481                    "1": "slot1",
 482                    "2": "slot2",
 483                    "3": "slot3",
 484                    "4": "slot4",
 485                    "5": "slot5",
 486                    "6": "slot6",
 487                    "7": "slot7",
 488                }),
 489
 490     # Check handling of default hash alg (sha256) with essiv
 491     LUKSConfig("aes-256-cbc-essiv-auto-sha1",
 492                "aes", 256, "cbc", "essiv", None, "sha1"),
 493
 494     # Check that a useless hash provided for 'plain64' iv gen
 495     # is ignored and no error raised
 496     LUKSConfig("aes-256-cbc-plain64-sha256-sha1",
 497                "aes", 256, "cbc", "plain64", "sha256", "sha1"),
 498
 499 ]
 500
 501 blacklist = [
 502     # We don't have a cast-6 cipher impl for QEMU yet
 503     "cast6-256-xts-plain64-sha1",
 504     "cast6-128-xts-plain64-sha1",
 505     "cast6-192-xts-plain64-sha1",
 506
 507     # GCrypt doesn't support Twofish with 192 bit key
 508     "twofish-192-xts-plain64-sha1",
 509 ]
 510
 511 whitelist = []
 512 if "LUKS_CONFIG" in os.environ:
 513     whitelist = os.environ["LUKS_CONFIG"].split(",")
 514
 515 for config in configs:
 516     if config.name in blacklist:
 517         iotests.log("Skipping %s in blacklist" % config.name)
 518         continue
 519
 520     if len(whitelist) > 0 and config.name not in whitelist:
 521         iotests.log("Skipping %s not in whitelist" % config.name)
 522         continue
 523
 524     test_once(config, qemu_img=False)
 525
 526     # XXX we should support setting passwords in a non-0
 527     # key slot with 'qemu-img create' in future
 528     (pw, slot) = config.first_password()
 529     if slot == "0":
 530         test_once(config, qemu_img=True)