mirror of https://gitlab.com/qemu-project/qemu
You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
553 lines
18 KiB
Python
553 lines
18 KiB
Python
#!/usr/bin/env python3
|
|
# group: rw sudo
|
|
#
|
|
# Copyright (C) 2016 Red Hat, Inc.
|
|
#
|
|
# This program is free software; you can redistribute it and/or modify
|
|
# it under the terms of the GNU General Public License as published by
|
|
# the Free Software Foundation; either version 2 of the License, or
|
|
# (at your option) any later version.
|
|
#
|
|
# This program is distributed in the hope that it will be useful,
|
|
# but WITHOUT ANY WARRANTY; without even the implied warranty of
|
|
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
|
|
# GNU General Public License for more details.
|
|
#
|
|
# You should have received a copy of the GNU General Public License
|
|
# along with this program. If not, see <http://www.gnu.org/licenses/>.
|
|
#
|
|
# Creator/Owner: Daniel P. Berrange <berrange@redhat.com>
|
|
#
|
|
# Exercise the QEMU 'luks' block driver to validate interoperability
|
|
# with the Linux dm-crypt + cryptsetup implementation
|
|
|
|
import subprocess
|
|
import os
|
|
import os.path
|
|
|
|
import base64
|
|
|
|
import iotests
|
|
|
|
|
|
class LUKSConfig(object):
|
|
"""Represent configuration parameters for a single LUKS
|
|
setup to be tested"""
|
|
|
|
def __init__(self, name, cipher, keylen, mode, ivgen,
|
|
ivgen_hash, hash, password=None, passwords=None):
|
|
|
|
self.name = name
|
|
self.cipher = cipher
|
|
self.keylen = keylen
|
|
self.mode = mode
|
|
self.ivgen = ivgen
|
|
self.ivgen_hash = ivgen_hash
|
|
self.hash = hash
|
|
|
|
if passwords is not None:
|
|
self.passwords = passwords
|
|
else:
|
|
self.passwords = {}
|
|
|
|
if password is None:
|
|
self.passwords["0"] = "123456"
|
|
else:
|
|
self.passwords["0"] = password
|
|
|
|
def __repr__(self):
|
|
return self.name
|
|
|
|
def image_name(self):
|
|
return "luks-%s.img" % self.name
|
|
|
|
def image_path(self):
|
|
return os.path.join(iotests.test_dir, self.image_name())
|
|
|
|
def device_name(self):
|
|
return "qiotest-145-%s" % self.name
|
|
|
|
def device_path(self):
|
|
return "/dev/mapper/" + self.device_name()
|
|
|
|
def first_password(self):
|
|
for i in range(8):
|
|
slot = str(i)
|
|
if slot in self.passwords:
|
|
return (self.passwords[slot], slot)
|
|
raise Exception("No password found")
|
|
|
|
def first_password_base64(self):
|
|
(pw, slot) = self.first_password()
|
|
return base64.b64encode(pw.encode('ascii')).decode('ascii')
|
|
|
|
def active_slots(self):
|
|
slots = []
|
|
for i in range(8):
|
|
slot = str(i)
|
|
if slot in self.passwords:
|
|
slots.append(slot)
|
|
return slots
|
|
|
|
def verify_passwordless_sudo():
|
|
"""Check whether sudo is configured to allow
|
|
password-less access to commands"""
|
|
|
|
args = ["sudo", "-n", "/bin/true"]
|
|
|
|
proc = subprocess.Popen(args,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
|
|
msg = proc.communicate()[0]
|
|
|
|
if proc.returncode != 0:
|
|
iotests.notrun('requires password-less sudo access: %s' % msg)
|
|
|
|
|
|
def cryptsetup(args, password=None):
|
|
"""Run the cryptsetup command in batch mode"""
|
|
|
|
fullargs = ["sudo", "cryptsetup", "-q", "-v"]
|
|
fullargs.extend(args)
|
|
|
|
iotests.log(" ".join(fullargs), filters=[iotests.filter_test_dir])
|
|
proc = subprocess.Popen(fullargs,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
universal_newlines=True)
|
|
|
|
msg = proc.communicate(password)[0]
|
|
|
|
if proc.returncode != 0:
|
|
raise Exception(msg)
|
|
|
|
|
|
def cryptsetup_add_password(config, slot):
|
|
"""Add another password to a LUKS key slot"""
|
|
|
|
(password, mainslot) = config.first_password()
|
|
|
|
pwfile = os.path.join(iotests.test_dir, "passwd.txt")
|
|
with open(pwfile, "w") as fh:
|
|
fh.write(config.passwords[slot])
|
|
|
|
try:
|
|
args = ["luksAddKey", config.image_path(),
|
|
"--key-slot", slot,
|
|
"--key-file", "-",
|
|
"--iter-time", "10",
|
|
pwfile]
|
|
|
|
cryptsetup(args, password)
|
|
finally:
|
|
os.unlink(pwfile)
|
|
|
|
|
|
def cryptsetup_format(config):
|
|
"""Format a new LUKS volume with cryptsetup, adding the
|
|
first key slot only"""
|
|
|
|
(password, slot) = config.first_password()
|
|
|
|
args = ["luksFormat", "--type", "luks1"]
|
|
cipher = config.cipher + "-" + config.mode + "-" + config.ivgen
|
|
if config.ivgen_hash is not None:
|
|
cipher = cipher + ":" + config.ivgen_hash
|
|
elif config.ivgen == "essiv":
|
|
cipher = cipher + ":" + "sha256"
|
|
args.extend(["--cipher", cipher])
|
|
if config.mode == "xts":
|
|
args.extend(["--key-size", str(config.keylen * 2)])
|
|
else:
|
|
args.extend(["--key-size", str(config.keylen)])
|
|
if config.hash is not None:
|
|
args.extend(["--hash", config.hash])
|
|
args.extend(["--key-slot", slot])
|
|
args.extend(["--key-file", "-"])
|
|
args.extend(["--iter-time", "10"])
|
|
args.append(config.image_path())
|
|
|
|
cryptsetup(args, password)
|
|
|
|
|
|
def chown(config):
|
|
"""Set the ownership of a open LUKS device to this user"""
|
|
|
|
path = config.device_path()
|
|
|
|
args = ["sudo", "chown", "%d:%d" % (os.getuid(), os.getgid()), path]
|
|
iotests.log(" ".join(args), filters=[iotests.filter_chown])
|
|
proc = subprocess.Popen(args,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT)
|
|
|
|
msg = proc.communicate()[0]
|
|
|
|
if proc.returncode != 0:
|
|
raise Exception(msg)
|
|
|
|
|
|
def cryptsetup_open(config):
|
|
"""Open an image as a LUKS device"""
|
|
|
|
(password, slot) = config.first_password()
|
|
|
|
args = ["luksOpen", config.image_path(), config.device_name()]
|
|
|
|
cryptsetup(args, password)
|
|
|
|
|
|
def cryptsetup_close(config):
|
|
"""Close an active LUKS device """
|
|
|
|
args = ["luksClose", config.device_name()]
|
|
cryptsetup(args)
|
|
|
|
|
|
def delete_image(config):
|
|
"""Delete a disk image"""
|
|
|
|
try:
|
|
os.unlink(config.image_path())
|
|
iotests.log("unlink %s" % config.image_path(),
|
|
filters=[iotests.filter_test_dir])
|
|
except Exception as e:
|
|
pass
|
|
|
|
|
|
def create_image(config, size_mb):
|
|
"""Create a bare disk image with requested size"""
|
|
|
|
delete_image(config)
|
|
iotests.log("truncate %s --size %dMB" % (config.image_path(), size_mb),
|
|
filters=[iotests.filter_test_dir])
|
|
with open(config.image_path(), "w") as fn:
|
|
fn.truncate(size_mb * 1024 * 1024)
|
|
|
|
|
|
def check_cipher_support(config, output):
|
|
"""Check the output of qemu-img or qemu-io for mention of the respective
|
|
cipher algorithm being unsupported, and if so, skip this test.
|
|
(Returns `output` for convenience.)"""
|
|
|
|
if 'Unsupported cipher algorithm' in output:
|
|
iotests.notrun('Unsupported cipher algorithm '
|
|
f'{config.cipher}-{config.keylen}-{config.mode}; '
|
|
'consider configuring qemu with a different crypto '
|
|
'backend')
|
|
return output
|
|
|
|
def qemu_img_create(config, size_mb):
|
|
"""Create and format a disk image with LUKS using qemu-img"""
|
|
|
|
opts = [
|
|
"key-secret=sec0",
|
|
"iter-time=10",
|
|
"cipher-alg=%s-%d" % (config.cipher, config.keylen),
|
|
"cipher-mode=%s" % config.mode,
|
|
"ivgen-alg=%s" % config.ivgen,
|
|
"hash-alg=%s" % config.hash,
|
|
]
|
|
if config.ivgen_hash is not None:
|
|
opts.append("ivgen-hash-alg=%s" % config.ivgen_hash)
|
|
|
|
args = ["create", "-f", "luks",
|
|
"--object",
|
|
("secret,id=sec0,data=%s,format=base64" %
|
|
config.first_password_base64()),
|
|
"-o", ",".join(opts),
|
|
config.image_path(),
|
|
"%dM" % size_mb]
|
|
|
|
iotests.log("qemu-img " + " ".join(args), filters=[iotests.filter_test_dir])
|
|
try:
|
|
iotests.qemu_img(*args)
|
|
except subprocess.CalledProcessError as exc:
|
|
check_cipher_support(config, exc.output)
|
|
raise
|
|
|
|
def qemu_io_image_args(config, dev=False):
|
|
"""Get the args for access an image or device with qemu-io"""
|
|
|
|
if dev:
|
|
return [
|
|
"--image-opts",
|
|
"driver=host_device,filename=%s" % config.device_path()]
|
|
else:
|
|
return [
|
|
"--object",
|
|
("secret,id=sec0,data=%s,format=base64" %
|
|
config.first_password_base64()),
|
|
"--image-opts",
|
|
("driver=luks,key-secret=sec0,file.filename=%s" %
|
|
config.image_path())]
|
|
|
|
def qemu_io_write_pattern(config, pattern, offset_mb, size_mb, dev=False):
|
|
"""Write a pattern of data to a LUKS image or device"""
|
|
|
|
if dev:
|
|
chown(config)
|
|
args = ["-c", "write -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)]
|
|
args.extend(qemu_io_image_args(config, dev))
|
|
iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir])
|
|
output = iotests.qemu_io(*args, check=False).stdout
|
|
iotests.log(check_cipher_support(config, output),
|
|
filters=[iotests.filter_test_dir, iotests.filter_qemu_io])
|
|
|
|
|
|
def qemu_io_read_pattern(config, pattern, offset_mb, size_mb, dev=False):
|
|
"""Read a pattern of data to a LUKS image or device"""
|
|
|
|
if dev:
|
|
chown(config)
|
|
args = ["-c", "read -P 0x%x %dM %dM" % (pattern, offset_mb, size_mb)]
|
|
args.extend(qemu_io_image_args(config, dev))
|
|
iotests.log("qemu-io " + " ".join(args), filters=[iotests.filter_test_dir])
|
|
output = iotests.qemu_io(*args, check=False).stdout
|
|
iotests.log(check_cipher_support(config, output),
|
|
filters=[iotests.filter_test_dir, iotests.filter_qemu_io])
|
|
|
|
|
|
def test_once(config, qemu_img=False):
|
|
"""Run the test with a desired LUKS configuration. Can either
|
|
use qemu-img for creating the initial volume, or cryptsetup,
|
|
in order to test interoperability in both directions"""
|
|
|
|
iotests.log("# ================= %s %s =================" % (
|
|
"qemu-img" if qemu_img else "dm-crypt", config))
|
|
|
|
oneKB = 1024
|
|
oneMB = oneKB * 1024
|
|
oneGB = oneMB * 1024
|
|
oneTB = oneGB * 1024
|
|
|
|
# 4 TB, so that we pass the 32-bit sector number boundary.
|
|
# Important for testing correctness of some IV generators
|
|
# The files are sparse, so not actually using this much space
|
|
image_size = 4 * oneTB
|
|
if qemu_img:
|
|
iotests.log("# Create image")
|
|
qemu_img_create(config, image_size // oneMB)
|
|
else:
|
|
iotests.log("# Create image")
|
|
create_image(config, image_size // oneMB)
|
|
|
|
lowOffsetMB = 100
|
|
highOffsetMB = 3 * oneTB // oneMB
|
|
|
|
try:
|
|
if not qemu_img:
|
|
iotests.log("# Format image")
|
|
cryptsetup_format(config)
|
|
|
|
for slot in config.active_slots()[1:]:
|
|
iotests.log("# Add password slot %s" % slot)
|
|
cryptsetup_add_password(config, slot)
|
|
|
|
# First we'll open the image using cryptsetup and write a
|
|
# known pattern of data that we'll then verify with QEMU
|
|
|
|
iotests.log("# Open dev")
|
|
cryptsetup_open(config)
|
|
|
|
try:
|
|
iotests.log("# Write test pattern 0xa7")
|
|
qemu_io_write_pattern(config, 0xa7, lowOffsetMB, 10, dev=True)
|
|
iotests.log("# Write test pattern 0x13")
|
|
qemu_io_write_pattern(config, 0x13, highOffsetMB, 10, dev=True)
|
|
finally:
|
|
iotests.log("# Close dev")
|
|
cryptsetup_close(config)
|
|
|
|
# Ok, now we're using QEMU to verify the pattern just
|
|
# written via dm-crypt
|
|
|
|
iotests.log("# Read test pattern 0xa7")
|
|
qemu_io_read_pattern(config, 0xa7, lowOffsetMB, 10, dev=False)
|
|
iotests.log("# Read test pattern 0x13")
|
|
qemu_io_read_pattern(config, 0x13, highOffsetMB, 10, dev=False)
|
|
|
|
|
|
# Write a new pattern to the image, which we'll later
|
|
# verify with dm-crypt
|
|
iotests.log("# Write test pattern 0x91")
|
|
qemu_io_write_pattern(config, 0x91, lowOffsetMB, 10, dev=False)
|
|
iotests.log("# Write test pattern 0x5e")
|
|
qemu_io_write_pattern(config, 0x5e, highOffsetMB, 10, dev=False)
|
|
|
|
|
|
# Now we're opening the image with dm-crypt once more
|
|
# and verifying what QEMU wrote, completing the circle
|
|
iotests.log("# Open dev")
|
|
cryptsetup_open(config)
|
|
|
|
try:
|
|
iotests.log("# Read test pattern 0x91")
|
|
qemu_io_read_pattern(config, 0x91, lowOffsetMB, 10, dev=True)
|
|
iotests.log("# Read test pattern 0x5e")
|
|
qemu_io_read_pattern(config, 0x5e, highOffsetMB, 10, dev=True)
|
|
finally:
|
|
iotests.log("# Close dev")
|
|
cryptsetup_close(config)
|
|
finally:
|
|
iotests.log("# Delete image")
|
|
delete_image(config)
|
|
print()
|
|
|
|
|
|
# Obviously we only work with the luks image format
|
|
iotests.script_initialize(supported_fmts=['luks'])
|
|
|
|
# We need sudo in order to run cryptsetup to create
|
|
# dm-crypt devices. This is safe to use on any
|
|
# machine, since all dm-crypt devices are backed
|
|
# by newly created plain files, and have a dm-crypt
|
|
# name prefix of 'qiotest' to avoid clashing with
|
|
# user LUKS volumes
|
|
verify_passwordless_sudo()
|
|
|
|
|
|
# If we look at all permutations of cipher, key size,
|
|
# mode, ivgen, hash, there are ~1000 possible configs.
|
|
#
|
|
# We certainly don't want/need to test every permutation
|
|
# to get good validation of interoperability between QEMU
|
|
# and dm-crypt/cryptsetup.
|
|
#
|
|
# The configs below are a representative set that aim to
|
|
# exercise each axis of configurability.
|
|
#
|
|
configs = [
|
|
# A common LUKS default
|
|
LUKSConfig("aes-256-xts-plain64-sha1",
|
|
"aes", 256, "xts", "plain64", None, "sha1"),
|
|
|
|
|
|
# LUKS default but diff ciphers
|
|
LUKSConfig("twofish-256-xts-plain64-sha1",
|
|
"twofish", 256, "xts", "plain64", None, "sha1"),
|
|
LUKSConfig("serpent-256-xts-plain64-sha1",
|
|
"serpent", 256, "xts", "plain64", None, "sha1"),
|
|
# Should really be xts, but kernel doesn't support xts+cast5
|
|
# nor does it do essiv+cast5
|
|
LUKSConfig("cast5-128-cbc-plain64-sha1",
|
|
"cast5", 128, "cbc", "plain64", None, "sha1"),
|
|
LUKSConfig("cast6-256-xts-plain64-sha1",
|
|
"cast6", 256, "xts", "plain64", None, "sha1"),
|
|
|
|
|
|
# LUKS default but diff modes / ivgens
|
|
LUKSConfig("aes-256-cbc-plain-sha1",
|
|
"aes", 256, "cbc", "plain", None, "sha1"),
|
|
LUKSConfig("aes-256-cbc-plain64-sha1",
|
|
"aes", 256, "cbc", "plain64", None, "sha1"),
|
|
LUKSConfig("aes-256-cbc-essiv-sha256-sha1",
|
|
"aes", 256, "cbc", "essiv", "sha256", "sha1"),
|
|
LUKSConfig("aes-256-xts-essiv-sha256-sha1",
|
|
"aes", 256, "xts", "essiv", "sha256", "sha1"),
|
|
|
|
|
|
# LUKS default but smaller key sizes
|
|
LUKSConfig("aes-128-xts-plain64-sha256-sha1",
|
|
"aes", 128, "xts", "plain64", None, "sha1"),
|
|
LUKSConfig("aes-192-xts-plain64-sha256-sha1",
|
|
"aes", 192, "xts", "plain64", None, "sha1"),
|
|
|
|
LUKSConfig("twofish-128-xts-plain64-sha1",
|
|
"twofish", 128, "xts", "plain64", None, "sha1"),
|
|
LUKSConfig("twofish-192-xts-plain64-sha1",
|
|
"twofish", 192, "xts", "plain64", None, "sha1"),
|
|
|
|
LUKSConfig("serpent-128-xts-plain64-sha1",
|
|
"serpent", 128, "xts", "plain64", None, "sha1"),
|
|
LUKSConfig("serpent-192-xts-plain64-sha1",
|
|
"serpent", 192, "xts", "plain64", None, "sha1"),
|
|
|
|
LUKSConfig("cast6-128-xts-plain64-sha1",
|
|
"cast6", 128, "xts", "plain", None, "sha1"),
|
|
LUKSConfig("cast6-192-xts-plain64-sha1",
|
|
"cast6", 192, "xts", "plain64", None, "sha1"),
|
|
|
|
|
|
# LUKS default but diff hash
|
|
LUKSConfig("aes-256-xts-plain64-sha224",
|
|
"aes", 256, "xts", "plain64", None, "sha224"),
|
|
LUKSConfig("aes-256-xts-plain64-sha256",
|
|
"aes", 256, "xts", "plain64", None, "sha256"),
|
|
LUKSConfig("aes-256-xts-plain64-sha384",
|
|
"aes", 256, "xts", "plain64", None, "sha384"),
|
|
LUKSConfig("aes-256-xts-plain64-sha512",
|
|
"aes", 256, "xts", "plain64", None, "sha512"),
|
|
LUKSConfig("aes-256-xts-plain64-ripemd160",
|
|
"aes", 256, "xts", "plain64", None, "ripemd160"),
|
|
|
|
# Password in slot 3
|
|
LUKSConfig("aes-256-xts-plain-sha1-pwslot3",
|
|
"aes", 256, "xts", "plain", None, "sha1",
|
|
passwords={
|
|
"3": "slot3",
|
|
}),
|
|
|
|
# Passwords in every slot
|
|
LUKSConfig("aes-256-xts-plain-sha1-pwallslots",
|
|
"aes", 256, "xts", "plain", None, "sha1",
|
|
passwords={
|
|
"0": "slot1",
|
|
"1": "slot1",
|
|
"2": "slot2",
|
|
"3": "slot3",
|
|
"4": "slot4",
|
|
"5": "slot5",
|
|
"6": "slot6",
|
|
"7": "slot7",
|
|
}),
|
|
|
|
# Check handling of default hash alg (sha256) with essiv
|
|
LUKSConfig("aes-256-cbc-essiv-auto-sha1",
|
|
"aes", 256, "cbc", "essiv", None, "sha1"),
|
|
|
|
# Check that a useless hash provided for 'plain64' iv gen
|
|
# is ignored and no error raised
|
|
LUKSConfig("aes-256-cbc-plain64-sha256-sha1",
|
|
"aes", 256, "cbc", "plain64", "sha256", "sha1"),
|
|
|
|
]
|
|
|
|
unsupported_configs = [
|
|
# We don't have a cast-6 cipher impl for QEMU yet
|
|
"cast6-256-xts-plain64-sha1",
|
|
"cast6-128-xts-plain64-sha1",
|
|
"cast6-192-xts-plain64-sha1",
|
|
|
|
# GCrypt doesn't support Twofish with 192 bit key
|
|
"twofish-192-xts-plain64-sha1",
|
|
]
|
|
|
|
# Optionally test only the configurations in the LUKS_CONFIG
|
|
# environment variable
|
|
tested_configs = None
|
|
if "LUKS_CONFIG" in os.environ:
|
|
tested_configs = os.environ["LUKS_CONFIG"].split(",")
|
|
|
|
for config in configs:
|
|
if config.name in unsupported_configs:
|
|
iotests.log("Skipping %s (config not supported)" % config.name)
|
|
continue
|
|
|
|
if tested_configs is not None and config.name not in tested_configs:
|
|
iotests.log("Skipping %s (by user request)" % config.name)
|
|
continue
|
|
|
|
test_once(config, qemu_img=False)
|
|
|
|
# XXX we should support setting passwords in a non-0
|
|
# key slot with 'qemu-img create' in future
|
|
(pw, slot) = config.first_password()
|
|
if slot == "0":
|
|
test_once(config, qemu_img=True)
|