Merge tag 'qemu-macppc-20230206' of https://github.com/mcayland/qemu into staging
[qemu.git] / tests / qemu-iotests / 149
blob2ae318f16f010b2d9b6864a1a8e05f4cb675f595
1 #!/usr/bin/env python3
2 # group: rw sudo
4 # Copyright (C) 2016 Red Hat, Inc.
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.
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.
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/>.
19 # Creator/Owner: Daniel P. Berrange <berrange@redhat.com>
21 # Exercise the QEMU 'luks' block driver to validate interoperability
22 # with the Linux dm-crypt + cryptsetup implementation
24 import subprocess
25 import os
26 import os.path
28 import base64
30 import iotests
33 class LUKSConfig(object):
34     """Represent configuration parameters for a single LUKS
35        setup to be tested"""
37     def __init__(self, name, cipher, keylen, mode, ivgen,
38                  ivgen_hash, hash, password=None, passwords=None):
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
48         if passwords is not None:
49             self.passwords = passwords
50         else:
51             self.passwords = {}
53             if password is None:
54                 self.passwords["0"] = "123456"
55             else:
56                 self.passwords["0"] = password
58     def __repr__(self):
59         return self.name
61     def image_name(self):
62         return "luks-%s.img" % self.name
64     def image_path(self):
65         return os.path.join(iotests.test_dir, self.image_name())
67     def device_name(self):
68         return "qiotest-145-%s" % self.name
70     def device_path(self):
71         return "/dev/mapper/" + self.device_name()
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")
80     def first_password_base64(self):
81         (pw, slot) = self.first_password()
82         return base64.b64encode(pw.encode('ascii')).decode('ascii')
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
92 def verify_passwordless_sudo():
93     """Check whether sudo is configured to allow
94        password-less access to commands"""
96     args = ["sudo", "-n", "/bin/true"]
98     proc = subprocess.Popen(args,
99                             stdin=subprocess.PIPE,
100                             stdout=subprocess.PIPE,
101                             stderr=subprocess.STDOUT,
102                             universal_newlines=True)
104     msg = proc.communicate()[0]
106     if proc.returncode != 0:
107         iotests.notrun('requires password-less sudo access: %s' % msg)
110 def cryptsetup(args, password=None):
111     """Run the cryptsetup command in batch mode"""
113     fullargs = ["sudo", "cryptsetup", "-q", "-v"]
114     fullargs.extend(args)
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)
123     msg = proc.communicate(password)[0]
125     if proc.returncode != 0:
126         raise Exception(msg)
129 def cryptsetup_add_password(config, slot):
130     """Add another password to a LUKS key slot"""
132     (password, mainslot) = config.first_password()
134     pwfile = os.path.join(iotests.test_dir, "passwd.txt")
135     with open(pwfile, "w") as fh:
136         fh.write(config.passwords[slot])
138     try:
139         args = ["luksAddKey", config.image_path(),
140                 "--key-slot", slot,
141                 "--key-file", "-",
142                 "--iter-time", "10",
143                 pwfile]
145         cryptsetup(args, password)
146     finally:
147         os.unlink(pwfile)
150 def cryptsetup_format(config):
151     """Format a new LUKS volume with cryptsetup, adding the
152     first key slot only"""
154     (password, slot) = config.first_password()
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())
174     cryptsetup(args, password)
177 def chown(config):
178     """Set the ownership of a open LUKS device to this user"""
180     path = config.device_path()
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)
189     msg = proc.communicate()[0]
191     if proc.returncode != 0:
192         raise Exception(msg)
195 def cryptsetup_open(config):
196     """Open an image as a LUKS device"""
198     (password, slot) = config.first_password()
200     args = ["luksOpen", config.image_path(), config.device_name()]
202     cryptsetup(args, password)
205 def cryptsetup_close(config):
206     """Close an active LUKS device """
208     args = ["luksClose", config.device_name()]
209     cryptsetup(args)
212 def delete_image(config):
213     """Delete a disk image"""
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
223 def create_image(config, size_mb):
224     """Create a bare disk image with requested size"""
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)
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.)"""
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
245 def qemu_img_create(config, size_mb):
246     """Create and format a disk image with LUKS using qemu-img"""
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)
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]
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
274 def qemu_io_image_args(config, dev=False):
275     """Get the args for access an image or device with qemu-io"""
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())]
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"""
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])
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"""
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])
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"""
321     iotests.log("# ================= %s %s =================" % (
322         "qemu-img" if qemu_img else "dm-crypt", config))
324     oneKB = 1024
325     oneMB = oneKB * 1024
326     oneGB = oneMB * 1024
327     oneTB = oneGB * 1024
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)
340     lowOffsetMB = 100
341     highOffsetMB = 3 * oneTB // oneMB
343     try:
344         if not qemu_img:
345             iotests.log("# Format image")
346             cryptsetup_format(config)
348             for slot in config.active_slots()[1:]:
349                 iotests.log("# Add password slot %s" % slot)
350                 cryptsetup_add_password(config, slot)
352         # First we'll open the image using cryptsetup and write a
353         # known pattern of data that we'll then verify with QEMU
355         iotests.log("# Open dev")
356         cryptsetup_open(config)
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)
367         # Ok, now we're using QEMU to verify the pattern just
368         # written via dm-crypt
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)
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)
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)
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()
403 # Obviously we only work with the luks image format
404 iotests.script_initialize(supported_fmts=['luks'])
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()
415 # If we look at all permutations of cipher, key size,
416 # mode, ivgen, hash, there are ~1000 possible configs.
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.
422 # The configs below are a representative set that aim to
423 # exercise each axis of configurability.
425 configs = [
426     # A common LUKS default
427     LUKSConfig("aes-256-xts-plain64-sha1",
428                "aes", 256, "xts", "plain64", None, "sha1"),
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"),
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"),
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"),
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"),
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"),
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"),
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"),
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                }),
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                }),
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"),
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"),
521 blacklist = [
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",
527     # GCrypt doesn't support Twofish with 192 bit key
528     "twofish-192-xts-plain64-sha1",
531 whitelist = []
532 if "LUKS_CONFIG" in os.environ:
533     whitelist = os.environ["LUKS_CONFIG"].split(",")
535 for config in configs:
536     if config.name in blacklist:
537         iotests.log("Skipping %s in blacklist" % config.name)
538         continue
540     if len(whitelist) > 0 and config.name not in whitelist:
541         iotests.log("Skipping %s not in whitelist" % config.name)
542         continue
544     test_once(config, qemu_img=False)
546     # XXX we should support setting passwords in a non-0
547     # key slot with 'qemu-img create' in future
548     (pw, slot) = config.first_password()
549     if slot == "0":
550         test_once(config, qemu_img=True)