| #!/usr/bin/env python3 | |
| # group: rw | |
| # | |
| # Test case for encryption key management versus image sharing | |
| # | |
| # Copyright (C) 2019 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/>. | |
| # | |
| import iotests | |
| import os | |
| import time | |
| import json | |
| test_img = os.path.join(iotests.test_dir, 'test.img') | |
| class Secret: | |
| def __init__(self, index): | |
| self._id = "keysec" + str(index) | |
| # you are not supposed to see the password... | |
| self._secret = "hunter" + str(index) | |
| def id(self): | |
| return self._id | |
| def secret(self): | |
| return self._secret | |
| def to_cmdline_object(self): | |
| return [ "secret,id=" + self._id + ",data=" + self._secret] | |
| def to_qmp_object(self): | |
| return { "qom-type" : "secret", "id": self.id(), | |
| "data": self.secret() } | |
| ################################################################################ | |
| class EncryptionSetupTestCase(iotests.QMPTestCase): | |
| # test case startup | |
| def setUp(self): | |
| # start the VMs | |
| self.vm1 = iotests.VM(path_suffix = 'VM1') | |
| self.vm2 = iotests.VM(path_suffix = 'VM2') | |
| self.vm1.launch() | |
| self.vm2.launch() | |
| # create the secrets and load 'em into the VMs | |
| self.secrets = [ Secret(i) for i in range(0, 4) ] | |
| for secret in self.secrets: | |
| self.vm1.cmd("object-add", secret.to_qmp_object()) | |
| self.vm2.cmd("object-add", secret.to_qmp_object()) | |
| # test case shutdown | |
| def tearDown(self): | |
| # stop the VM | |
| self.vm1.shutdown() | |
| self.vm2.shutdown() | |
| ########################################################################### | |
| # create the encrypted block device using qemu-img | |
| def createImg(self, file, secret): | |
| iotests.qemu_img( | |
| 'create', | |
| '--object', *secret.to_cmdline_object(), | |
| '-f', iotests.imgfmt, | |
| '-o', 'key-secret=' + secret.id(), | |
| '-o', 'iter-time=10', | |
| file, | |
| '1M') | |
| iotests.log('') | |
| # attempts to add a key using qemu-img | |
| def addKey(self, file, secret, new_secret): | |
| image_options = { | |
| 'key-secret' : secret.id(), | |
| 'driver' : iotests.imgfmt, | |
| 'file' : { | |
| 'driver':'file', | |
| 'filename': file, | |
| } | |
| } | |
| output = iotests.qemu_img( | |
| 'amend', | |
| '--object', *secret.to_cmdline_object(), | |
| '--object', *new_secret.to_cmdline_object(), | |
| '-o', 'state=active', | |
| '-o', 'new-secret=' + new_secret.id(), | |
| '-o', 'iter-time=10', | |
| "json:" + json.dumps(image_options), | |
| check=False # Expected to fail. Log output. | |
| ).stdout | |
| iotests.log(output, filters=[iotests.filter_test_dir]) | |
| ########################################################################### | |
| # open an encrypted block device | |
| def openImageQmp(self, vm, id, file, secret, | |
| readOnly = False, reOpen = False): | |
| command = 'blockdev-reopen' if reOpen else 'blockdev-add' | |
| opts = { | |
| 'driver': iotests.imgfmt, | |
| 'node-name': id, | |
| 'read-only': readOnly, | |
| 'key-secret' : secret.id(), | |
| 'file': { | |
| 'driver': 'file', | |
| 'filename': test_img, | |
| } | |
| } | |
| if reOpen: | |
| vm.cmd(command, options=[opts]) | |
| else: | |
| vm.cmd(command, opts) | |
| ########################################################################### | |
| # add virtio-blk consumer for a block device | |
| def addImageUser(self, vm, id, disk_id, share_rw=False): | |
| result = vm.qmp('device_add', { | |
| 'driver': 'virtio-blk', | |
| 'id': id, | |
| 'drive': disk_id, | |
| 'share-rw' : share_rw | |
| } | |
| ) | |
| iotests.log(result) | |
| # close the encrypted block device | |
| def closeImageQmp(self, vm, id): | |
| vm.cmd('blockdev-del', {'node-name': id}) | |
| ########################################################################### | |
| # add a key to an encrypted block device | |
| def addKeyQmp(self, vm, id, new_secret): | |
| args = { | |
| 'node-name': id, | |
| 'job-id' : 'job0', | |
| 'options' : { | |
| 'state' : 'active', | |
| 'driver' : iotests.imgfmt, | |
| 'new-secret': new_secret.id(), | |
| 'iter-time' : 10 | |
| }, | |
| } | |
| result = vm.qmp('x-blockdev-amend', args) | |
| iotests.log(result) | |
| # Run the job only if it was created | |
| event = ('JOB_STATUS_CHANGE', | |
| {'data': {'id': 'job0', 'status': 'created'}}) | |
| if vm.events_wait([event], timeout=0.0) is not None: | |
| vm.run_job('job0') | |
| # test that when the image opened by two qemu processes, | |
| # neither of them can update the encryption keys | |
| def test1(self): | |
| self.createImg(test_img, self.secrets[0]); | |
| # VM1 opens the image and adds a key | |
| self.openImageQmp(self.vm1, "testdev", test_img, self.secrets[0]) | |
| self.addKeyQmp(self.vm1, "testdev", new_secret = self.secrets[1]) | |
| # VM2 opens the image | |
| self.openImageQmp(self.vm2, "testdev", test_img, self.secrets[0]) | |
| # neither VMs now should be able to add a key | |
| self.addKeyQmp(self.vm1, "testdev", new_secret = self.secrets[2]) | |
| self.addKeyQmp(self.vm2, "testdev", new_secret = self.secrets[2]) | |
| # VM 1 closes the image | |
| self.closeImageQmp(self.vm1, "testdev") | |
| # now VM2 can add the key | |
| self.addKeyQmp(self.vm2, "testdev", new_secret = self.secrets[2]) | |
| # qemu-img should also not be able to add a key | |
| self.addKey(test_img, self.secrets[0], self.secrets[2]) | |
| # cleanup | |
| self.closeImageQmp(self.vm2, "testdev") | |
| os.remove(test_img) | |
| # test that when the image opened by two qemu processes, | |
| # even if first VM opens it read-only, the second can't update encryption | |
| # keys | |
| def test2(self): | |
| self.createImg(test_img, self.secrets[0]); | |
| # VM1 opens the image readonly | |
| self.openImageQmp(self.vm1, "testdev", test_img, self.secrets[0], | |
| readOnly = True) | |
| # VM2 opens the image | |
| self.openImageQmp(self.vm2, "testdev", test_img, self.secrets[0]) | |
| # VM1 can't add a key since image is readonly | |
| self.addKeyQmp(self.vm1, "testdev", new_secret = self.secrets[2]) | |
| # VM2 can't add a key since VM is has the image opened | |
| self.addKeyQmp(self.vm2, "testdev", new_secret = self.secrets[2]) | |
| #VM1 reopens the image read-write | |
| self.openImageQmp(self.vm1, "testdev", test_img, self.secrets[0], | |
| reOpen = True, readOnly = False) | |
| # VM1 still can't add the key | |
| self.addKeyQmp(self.vm1, "testdev", new_secret = self.secrets[2]) | |
| # VM2 gets away | |
| self.closeImageQmp(self.vm2, "testdev") | |
| # VM1 now can add the key | |
| self.addKeyQmp(self.vm1, "testdev", new_secret = self.secrets[2]) | |
| self.closeImageQmp(self.vm1, "testdev") | |
| os.remove(test_img) | |
| # test that two VMs can't open the same luks image by default | |
| # and attach it to a guest device | |
| def test3(self): | |
| self.createImg(test_img, self.secrets[0]); | |
| self.openImageQmp(self.vm1, "testdev", test_img, self.secrets[0]) | |
| self.addImageUser(self.vm1, "testctrl", "testdev") | |
| self.openImageQmp(self.vm2, "testdev", test_img, self.secrets[0]) | |
| self.addImageUser(self.vm2, "testctrl", "testdev") | |
| # test that two VMs can attach the same luks image to a guest device, | |
| # if both use share-rw=on | |
| def test4(self): | |
| self.createImg(test_img, self.secrets[0]); | |
| self.openImageQmp(self.vm1, "testdev", test_img, self.secrets[0]) | |
| self.addImageUser(self.vm1, "testctrl", "testdev", share_rw=True) | |
| self.openImageQmp(self.vm2, "testdev", test_img, self.secrets[0]) | |
| self.addImageUser(self.vm2, "testctrl", "testdev", share_rw=True) | |
| if __name__ == '__main__': | |
| # support only raw luks since luks encrypted qcow2 is a proper | |
| # format driver which doesn't allow any sharing | |
| iotests.activate_logging() | |
| iotests.main(supported_fmts = ['luks']) |