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 <berrange@redhat.com>
  19 #
  20 # Exercise the QEMU 'luks' block driver to validate interoperability
  21 # with the Linux dm-crypt + cryptsetup implementation
  22
  23 from __future__ import print_function
  24 import subprocess
  25 import os
  26 import os.path
  27
  28 import base64
  29
  30 import iotests
  31
  32
  33 class 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
  92 def 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
 109 def 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
 127 def 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
 148 def 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
 175 def 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
 193 def 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
 203 def cryptsetup_close(config):
 204     """Close an active LUKS device """
 205
 206     args = ["luksClose", config.device_name()]
 207     cryptsetup(args)
 208
 209
 210 def 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
 221 def 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
 231 def 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
 256 def 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=host_device,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
 272 def 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
 284 def 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
 296 def 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
 384 iotests.verify_image_format(supported_fmts=['luks'])
 385 iotests.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
 393 verify_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 #
 406 configs = [
 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
 502 blacklist = [
 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
 512 whitelist = []
 513 if "LUKS_CONFIG" in os.environ:
 514     whitelist = os.environ["LUKS_CONFIG"].split(",")
 515
 516 for 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)