xref: /openbmc/qemu/tests/qemu-iotests/149 (revision 502d0f36) 
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
23import subprocess
24import os
25import os.path
26
27import base64
28
29import iotests
30
31
32class 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
91def 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
108def 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
126def 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
147def 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
174def 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
192def 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
202def cryptsetup_close(config):
203    """Close an active LUKS device """
204
205    args = ["luksClose", config.device_name()]
206    cryptsetup(args)
207
208
209def 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
220def 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
230def 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
255def 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
271def 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
283def 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
295def 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
383iotests.verify_image_format(supported_fmts=['luks'])
384iotests.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
392verify_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#
405configs = [
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
501blacklist = [
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
511whitelist = []
512if "LUKS_CONFIG" in os.environ:
513    whitelist = os.environ["LUKS_CONFIG"].split(",")
514
515for 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)
531