xref: /openbmc/qemu/tests/qemu-iotests/149 (revision effd60c8) 
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
24import subprocess
25import os
26import os.path
27
28import base64
29
30import iotests
31
32
33class 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
92def 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
110def 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
129def 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
150def 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
177def 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
195def 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
205def cryptsetup_close(config):
206    """Close an active LUKS device """
207
208    args = ["luksClose", config.device_name()]
209    cryptsetup(args)
210
211
212def 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
223def 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
233def 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
245def 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
274def 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
290def 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
303def 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
316def 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
404iotests.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
412verify_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#
425configs = [
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
521unsupported_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
533tested_configs = None
534if "LUKS_CONFIG" in os.environ:
535    tested_configs = os.environ["LUKS_CONFIG"].split(",")
536
537for 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)
553