openstack/nova
Code Issues Proposed changes

encryptors: Workaround mangled passphrases

Browse Source 
Prior to Ib563b0ea the passphrase used by CryptsetupEncryptor and
LuksEncryptor had any leading zeros per hexadecimal digit removed, for
example 0x04 or 04 would turn into 0x4 or 4. As a result any volume
encrypted prior to the release of Newton used a modified passphrase that
was different to that stored by the key manager being used in the
environment.

To correct this for LuksEncryptor volumes permission denied errors are
now caught when attempting to open a volume. A second attempt to open
the volume is then made using a mangled passphrase. If successful the
correct passphrase is then added to the volume before the mangled
passphrase is finally removed. This workaround can be removed in a
future release once it is safe to assume that all LuksEncryptor volumes
have had any mangled passphrases replaced in this way.

This isn't possible for CryptsetupEncryptor volumes as the plain mode
used by cryptsetup does not provide a way for adding and removing keys.
As such on a permission denied error a second attempt is made to open
the volume using a mangled passphrase. Unlike the above workaround this
cannot be removed in a future release.

Change-Id: I7096463c5eba951dd6322ee6965435e877ca0371
Partial-bug: #1633518
This commit is contained in:
Lee Yarwood 2016-10-14 16:41:53 +01:00
parent ad695a2691
commit 2010c13c0a
4 changed files with 182 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

40
nova/tests/unit/volume/encryptors/test_cryptsetup.py
View File

@ -19,16 +19,17 @@ import copy
from castellan.common.objects import symmetric_key as key
import mock
from oslo_concurrency import processutils
import six
import uuid
from nova import exception
from nova.tests.unit.volume.encryptors import test_base
from nova.volume.encryptors import cryptsetup
def fake__get_key(context):
raw = bytes(binascii.unhexlify('0' * 32))
def fake__get_key(context, passphrase):
raw = bytes(binascii.unhexlify(passphrase))
symmetric_key = key.SymmetricKey('AES', len(raw) * 8, raw)
return symmetric_key
@ -59,14 +60,15 @@ class CryptsetupEncryptorTestCase(test_base.VolumeEncryptorTestCase):
@mock.patch('nova.utils.execute')
def test_attach_volume(self, mock_execute):
fake_key = uuid.uuid4().hex
self.encryptor._get_key = mock.MagicMock()
self.encryptor._get_key.return_value = fake__get_key(None)
self.encryptor._get_key.return_value = fake__get_key(None, fake_key)
self.encryptor.attach_volume(None)
mock_execute.assert_has_calls([
mock.call('cryptsetup', 'create', '--key-file=-', self.dev_name,
self.dev_path, process_input='0' * 32,
self.dev_path, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('ln', '--symbolic', '--force',
'/dev/mapper/%s' % self.dev_name, self.symlink_path,
@ -138,3 +140,31 @@ class CryptsetupEncryptorTestCase(test_base.VolumeEncryptorTestCase):
mock.call('/dev/mapper/%s' % wwn)])
mock_execute.assert_called_once_with(
'cryptsetup', 'status', wwn, run_as_root=True)
@mock.patch('nova.utils.execute')
def test_attach_volume_unmangle_passphrase(self, mock_execute):
fake_key = '0725230b'
fake_key_mangled = '72523b'
self.encryptor._get_key = mock.MagicMock()
self.encryptor._get_key.return_value = fake__get_key(None, fake_key)
mock_execute.side_effect = [
processutils.ProcessExecutionError(exit_code=2), # luksOpen
mock.DEFAULT,
mock.DEFAULT,
]
self.encryptor.attach_volume(None)
mock_execute.assert_has_calls([
mock.call('cryptsetup', 'create', '--key-file=-', self.dev_name,
self.dev_path, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', 'create', '--key-file=-', self.dev_name,
self.dev_path, process_input=fake_key_mangled,
run_as_root=True, check_exit_code=True),
mock.call('ln', '--symbolic', '--force',
'/dev/mapper/%s' % self.dev_name, self.symlink_path,
run_as_root=True, check_exit_code=True),
])
self.assertEqual(3, mock_execute.call_count)

91
nova/tests/unit/volume/encryptors/test_luks.py
View File

@ -14,8 +14,11 @@
# under the License.
import binascii
from castellan.common.objects import symmetric_key as key
import mock
from oslo_concurrency import processutils
import uuid
from nova.tests.unit.volume.encryptors import test_cryptsetup
from nova.volume.encryptors import luks
@ -78,15 +81,16 @@ class LuksEncryptorTestCase(test_cryptsetup.CryptsetupEncryptorTestCase):
@mock.patch('nova.utils.execute')
def test_attach_volume(self, mock_execute):
fake_key = uuid.uuid4().hex
self.encryptor._get_key = mock.MagicMock()
self.encryptor._get_key.return_value = \
test_cryptsetup.fake__get_key(None)
self.encryptor._get_key.return_value = test_cryptsetup.fake__get_key(
None, fake_key)
self.encryptor.attach_volume(None)
mock_execute.assert_has_calls([
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input='0' * 32,
self.dev_name, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('ln', '--symbolic', '--force',
'/dev/mapper/%s' % self.dev_name, self.symlink_path,
@ -96,9 +100,10 @@ class LuksEncryptorTestCase(test_cryptsetup.CryptsetupEncryptorTestCase):
@mock.patch('nova.utils.execute')
def test_attach_volume_not_formatted(self, mock_execute):
fake_key = uuid.uuid4().hex
self.encryptor._get_key = mock.MagicMock()
self.encryptor._get_key.return_value = \
test_cryptsetup.fake__get_key(None)
self.encryptor._get_key.return_value = test_cryptsetup.fake__get_key(
None, fake_key)
mock_execute.side_effect = [
processutils.ProcessExecutionError(exit_code=1), # luksOpen
@ -112,15 +117,15 @@ class LuksEncryptorTestCase(test_cryptsetup.CryptsetupEncryptorTestCase):
mock_execute.assert_has_calls([
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input='0' * 32,
self.dev_name, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', 'isLuks', '--verbose', self.dev_path,
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', '--batch-mode', 'luksFormat',
'--key-file=-', self.dev_path, process_input='0' * 32,
'--key-file=-', self.dev_path, process_input=fake_key,
run_as_root=True, check_exit_code=True, attempts=3),
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input='0' * 32,
self.dev_name, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('ln', '--symbolic', '--force',
'/dev/mapper/%s' % self.dev_name, self.symlink_path,
@ -130,9 +135,10 @@ class LuksEncryptorTestCase(test_cryptsetup.CryptsetupEncryptorTestCase):
@mock.patch('nova.utils.execute')
def test_attach_volume_fail(self, mock_execute):
fake_key = uuid.uuid4().hex
self.encryptor._get_key = mock.MagicMock()
self.encryptor._get_key.return_value = \
test_cryptsetup.fake__get_key(None)
self.encryptor._get_key.return_value = test_cryptsetup.fake__get_key(
None, fake_key)
mock_execute.side_effect = [
processutils.ProcessExecutionError(exit_code=1), # luksOpen
@ -144,7 +150,7 @@ class LuksEncryptorTestCase(test_cryptsetup.CryptsetupEncryptorTestCase):
mock_execute.assert_has_calls([
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input='0' * 32,
self.dev_name, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', 'isLuks', '--verbose', self.dev_path,
run_as_root=True, check_exit_code=True),
@ -170,3 +176,66 @@ class LuksEncryptorTestCase(test_cryptsetup.CryptsetupEncryptorTestCase):
attempts=3, run_as_root=True, check_exit_code=True),
])
self.assertEqual(1, mock_execute.call_count)
def test_get_mangled_passphrase(self):
# Confirm that a mangled passphrase is provided as per bug#1633518
unmangled_raw_key = bytes(binascii.unhexlify('0725230b'))
symmetric_key = key.SymmetricKey('AES', len(unmangled_raw_key) * 8,
unmangled_raw_key)
unmangled_encoded_key = symmetric_key.get_encoded()
encryptor = luks.LuksEncryptor(self.connection_info)
self.assertEqual(encryptor._get_mangled_passphrase(
unmangled_encoded_key), '72523b')
@mock.patch('nova.utils.execute')
def test_attach_volume_unmangle_passphrase(self, mock_execute):
fake_key = '0725230b'
fake_key_mangled = '72523b'
self.encryptor._get_key = mock.MagicMock(name='mock_execute')
self.encryptor._get_key.return_value = \
test_cryptsetup.fake__get_key(None, fake_key)
mock_execute.side_effect = [
processutils.ProcessExecutionError(exit_code=2), # luksOpen
mock.DEFAULT, # luksOpen
mock.DEFAULT, # luksClose
mock.DEFAULT, # luksAddKey
mock.DEFAULT, # luksOpen
mock.DEFAULT, # luksClose
mock.DEFAULT, # luksRemoveKey
mock.DEFAULT, # luksOpen
mock.DEFAULT, # ln
]
self.encryptor.attach_volume(None)
mock_execute.assert_has_calls([
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input=fake_key_mangled,
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', 'luksClose', self.dev_name,
run_as_root=True, check_exit_code=True, attempts=3),
mock.call('cryptsetup', 'luksAddKey', self.dev_path,
process_input=''.join([fake_key_mangled,
'\n', fake_key,
'\n', fake_key]),
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('cryptsetup', 'luksClose', self.dev_name,
run_as_root=True, check_exit_code=True, attempts=3),
mock.call('cryptsetup', 'luksRemoveKey', self.dev_path,
process_input=fake_key_mangled, run_as_root=True,
check_exit_code=True),
mock.call('cryptsetup', 'luksOpen', '--key-file=-', self.dev_path,
self.dev_name, process_input=fake_key,
run_as_root=True, check_exit_code=True),
mock.call('ln', '--symbolic', '--force',
'/dev/mapper/%s' % self.dev_name, self.symlink_path,
run_as_root=True, check_exit_code=True),
], any_order=False)
self.assertEqual(9, mock_execute.call_count)

25
nova/volume/encryptors/cryptsetup.py
View File

@ -14,6 +14,7 @@
# under the License.
import array
import binascii
import os
@ -21,7 +22,7 @@ from oslo_concurrency import processutils
from oslo_log import log as logging
from nova import exception
from nova.i18n import _LW
from nova.i18n import _LW, _LI
from nova import utils
from nova.volume.encryptors import base
@ -119,6 +120,17 @@ class CryptsetupEncryptor(base.VolumeEncryptor):
utils.execute(*cmd, process_input=passphrase,
check_exit_code=True, run_as_root=True)
def _get_mangled_passphrase(self, key):
"""Convert the raw key into a list of unsigned int's and then a string
"""
# NOTE(lyarwood): This replicates the methods used prior to Newton to
# first encode the passphrase as a list of unsigned int's before
# decoding back into a string. This method strips any leading 0's
# of the resulting hex digit pairs, resulting in a different
# passphrase being returned.
encoded_key = array.array('B', key).tolist()
return ''.join(hex(x).replace('0x', '') for x in encoded_key)
def attach_volume(self, context, **kwargs):
"""Shadows the device and passes an unencrypted version to the
instance.
@ -132,7 +144,16 @@ class CryptsetupEncryptor(base.VolumeEncryptor):
key = self._get_key(context).get_encoded()
passphrase = self._get_passphrase(key)
self._open_volume(passphrase, **kwargs)
try:
self._open_volume(passphrase, **kwargs)
except processutils.ProcessExecutionError as e:
if e.exit_code == 2:
# NOTE(lyarwood): Workaround bug#1633518 by attempting to use
# a mangled passphrase to open the device..
LOG.info(_LI("Unable to open %s with the current passphrase, "
"attempting to use a mangled passphrase to open "
"the volume."), self.dev_path)
self._open_volume(self._get_mangled_passphrase(key), **kwargs)
# modify the original symbolic link to refer to the decrypted device
utils.execute('ln', '--symbolic', '--force',

44
nova/volume/encryptors/luks.py
View File

@ -85,6 +85,40 @@ class LuksEncryptor(cryptsetup.CryptsetupEncryptor):
self.dev_path, self.dev_name, process_input=passphrase,
run_as_root=True, check_exit_code=True)
def _unmangle_volume(self, key, passphrase, **kwargs):
"""Workaround bug#1633518 by first identifying if a mangled passphrase
is used before replacing it with the correct passphrase.
"""
mangled_passphrase = self._get_mangled_passphrase(key)
self._open_volume(mangled_passphrase, **kwargs)
self._close_volume(**kwargs)
LOG.debug("%s correctly opened with a mangled passphrase, replacing"
"this with the original passphrase", self.dev_path)
# NOTE(lyarwood): Now that we are sure that the mangled passphrase is
# used attempt to add the correct passphrase before removing the
# mangled version from the volume.
# luksAddKey currently prompts for the following input :
# Enter any existing passphrase:
# Enter new passphrase for key slot:
# Verify passphrase:
utils.execute('cryptsetup', 'luksAddKey', self.dev_path,
process_input=''.join([mangled_passphrase, '\n',
passphrase, '\n', passphrase]),
run_as_root=True, check_exit_code=True)
# Verify that we can open the volume with the current passphrase
# before removing the mangled passphrase.
self._open_volume(passphrase, **kwargs)
self._close_volume(**kwargs)
# luksRemoveKey only prompts for the key to remove.
utils.execute('cryptsetup', 'luksRemoveKey', self.dev_path,
process_input=mangled_passphrase,
run_as_root=True, check_exit_code=True)
LOG.debug("%s mangled passphrase successfully replaced", self.dev_path)
def attach_volume(self, context, **kwargs):
"""Shadows the device and passes an unencrypted version to the
instance.
@ -108,6 +142,16 @@ class LuksEncryptor(cryptsetup.CryptsetupEncryptor):
self.dev_path)
self._format_volume(passphrase, **kwargs)
self._open_volume(passphrase, **kwargs)
elif e.exit_code == 2:
# NOTE(lyarwood): Workaround bug#1633518 by replacing any
# mangled passphrases that are found on the volume.
# TODO(lyarwood): Remove workaround during R.
LOG.warning(_LW("%s is not usable with the current "
"passphrase, attempting to use a mangled "
"passphrase to open the volume."),
self.dev_path)
self._unmangle_volume(key, passphrase, **kwargs)
self._open_volume(passphrase, **kwargs)
else:
raise