xref: /openbmc/qemu/tests/qemu-iotests/149 (revision 5eca450b) 
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
23from __future__ import print_function
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)
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
103    msg = proc.communicate()[0]
104
105    if proc.returncode != 0:
106        iotests.notrun('requires password-less sudo access: %s' % msg)
107
108
109def 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
127def 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
148def 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
175def 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
193def 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
203def cryptsetup_close(config):
204    """Close an active LUKS device """
205
206    args = ["luksClose", config.device_name()]
207    cryptsetup(args)
208
209
210def 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
221def 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
231def 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
256def 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=file,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
272def 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
284def 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
296def 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
384iotests.verify_image_format(supported_fmts=['luks'])
385iotests.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
393verify_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#
406configs = [
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
502blacklist = [
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
512whitelist = []
513if "LUKS_CONFIG" in os.environ:
514    whitelist = os.environ["LUKS_CONFIG"].split(",")
515
516for 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)
532