tests/qemu-iotests/149

   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
  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.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
  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                             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
 110 def 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
 129 def 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
 150 def 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
 177 def 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
 195 def 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
 205 def cryptsetup_close(config):
 206     """Close an active LUKS device """
 207
 208     args = ["luksClose", config.device_name()]
 209     cryptsetup(args)
 210
 211
 212 def 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
 223 def 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
 233 def 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
 245 def 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
 274 def 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
 290 def 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
 303 def 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
 316 def 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
 404 iotests.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
 412 verify_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 #
 425 configs = [
 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
 521 unsupported_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
 533 tested_configs = None
 534 if "LUKS_CONFIG" in os.environ:
 535     tested_configs = os.environ["LUKS_CONFIG"].split(",")
 536
 537 for 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)