Remove padding code and tests in the p11 plugin
Added new tox target "coverage" for generating html coverage. To run: tox -e coverage Change-Id: Ia60a0ed0749c27d9f9ee17058b0b027adc485e30
This commit is contained in:
parent
0cef83ff25
commit
affa4db238
@ -63,16 +63,6 @@ class P11CryptoPlugin(CryptoPluginBase):
|
|||||||
self.rw_session = self.pkcs11.openSession(1, PyKCS11.CKF_RW_SESSION)
|
self.rw_session = self.pkcs11.openSession(1, PyKCS11.CKF_RW_SESSION)
|
||||||
self.rw_session.login(conf.p11_crypto_plugin.login)
|
self.rw_session.login(conf.p11_crypto_plugin.login)
|
||||||
|
|
||||||
def _pad(self, unencrypted):
|
|
||||||
"""Adds padding to unencrypted byte string."""
|
|
||||||
pad_length = self.block_size - (len(unencrypted) % self.block_size)
|
|
||||||
return unencrypted + (chr(pad_length) * pad_length)
|
|
||||||
|
|
||||||
def _strip_pad(self, unencrypted):
|
|
||||||
pad_length = ord(unencrypted[-1:])
|
|
||||||
unpadded = unencrypted[:-pad_length]
|
|
||||||
return unpadded
|
|
||||||
|
|
||||||
def _check_error(self, value):
|
def _check_error(self, value):
|
||||||
if value != PyKCS11.CKR_OK:
|
if value != PyKCS11.CKR_OK:
|
||||||
# TODO: probably shouldn't raise PyKCS11 error here
|
# TODO: probably shouldn't raise PyKCS11 error here
|
||||||
@ -107,13 +97,11 @@ class P11CryptoPlugin(CryptoPluginBase):
|
|||||||
return gcm
|
return gcm
|
||||||
|
|
||||||
def encrypt(self, unencrypted, kek_meta_tenant, tenant):
|
def encrypt(self, unencrypted, kek_meta_tenant, tenant):
|
||||||
# TODO: GCM should not require padding.
|
|
||||||
padded_data = self._pad(unencrypted)
|
|
||||||
key = self._get_key_by_label(kek_meta_tenant.kek_label)
|
key = self._get_key_by_label(kek_meta_tenant.kek_label)
|
||||||
iv = self._generate_iv()
|
iv = self._generate_iv()
|
||||||
gcm = self._build_gcm_params(iv)
|
gcm = self._build_gcm_params(iv)
|
||||||
mech = PyKCS11.Mechanism(self.algorithm, gcm)
|
mech = PyKCS11.Mechanism(self.algorithm, gcm)
|
||||||
encrypted = self.session.encrypt(key, padded_data, mech)
|
encrypted = self.session.encrypt(key, unencrypted, mech)
|
||||||
cyphertext = b''.join(chr(i) for i in encrypted)
|
cyphertext = b''.join(chr(i) for i in encrypted)
|
||||||
kek_meta_extended = json.dumps({
|
kek_meta_extended = json.dumps({
|
||||||
'iv': base64.b64encode(iv)
|
'iv': base64.b64encode(iv)
|
||||||
@ -128,8 +116,8 @@ class P11CryptoPlugin(CryptoPluginBase):
|
|||||||
gcm = self._build_gcm_params(iv)
|
gcm = self._build_gcm_params(iv)
|
||||||
mech = PyKCS11.Mechanism(self.algorithm, gcm)
|
mech = PyKCS11.Mechanism(self.algorithm, gcm)
|
||||||
decrypted = self.session.decrypt(key, encrypted, mech)
|
decrypted = self.session.decrypt(key, encrypted, mech)
|
||||||
padded_secret = b''.join(chr(i) for i in decrypted)
|
secret = b''.join(chr(i) for i in decrypted)
|
||||||
return self._strip_pad(padded_secret)
|
return secret
|
||||||
|
|
||||||
def bind_kek_metadata(self, kek_metadata):
|
def bind_kek_metadata(self, kek_metadata):
|
||||||
# Enforce idempotency: If we've already generated a key for the
|
# Enforce idempotency: If we've already generated a key for the
|
||||||
|
@ -13,7 +13,6 @@
|
|||||||
# See the License for the specific language governing permissions and
|
# See the License for the specific language governing permissions and
|
||||||
# limitations under the License.
|
# limitations under the License.
|
||||||
|
|
||||||
from Crypto import Random
|
|
||||||
from mock import MagicMock
|
from mock import MagicMock
|
||||||
from mock import patch
|
from mock import patch
|
||||||
import unittest
|
import unittest
|
||||||
@ -41,32 +40,6 @@ class WhenTestingP11CryptoPlugin(unittest.TestCase):
|
|||||||
def tearDown(self):
|
def tearDown(self):
|
||||||
self.patcher.stop()
|
self.patcher.stop()
|
||||||
|
|
||||||
def test_pad_binary_string(self):
|
|
||||||
binary_string = b'some_binary_string'
|
|
||||||
padded_string = (
|
|
||||||
b'some_binary_string' +
|
|
||||||
b'\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e'
|
|
||||||
)
|
|
||||||
self.assertEqual(self.plugin._pad(binary_string), padded_string)
|
|
||||||
|
|
||||||
def test_pad_random_bytes(self):
|
|
||||||
random_bytes = Random.get_random_bytes(10)
|
|
||||||
padded_bytes = random_bytes + b'\x06\x06\x06\x06\x06\x06'
|
|
||||||
self.assertEqual(self.plugin._pad(random_bytes), padded_bytes)
|
|
||||||
|
|
||||||
def test_strip_padding_from_binary_string(self):
|
|
||||||
binary_string = b'some_binary_string'
|
|
||||||
padded_string = (
|
|
||||||
b'some_binary_string' +
|
|
||||||
b'\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e\x0e'
|
|
||||||
)
|
|
||||||
self.assertEqual(self.plugin._strip_pad(padded_string), binary_string)
|
|
||||||
|
|
||||||
def test_strip_padding_from_random_bytes(self):
|
|
||||||
random_bytes = Random.get_random_bytes(10)
|
|
||||||
padded_bytes = random_bytes + b'\x06\x06\x06\x06\x06\x06'
|
|
||||||
self.assertEqual(self.plugin._strip_pad(padded_bytes), random_bytes)
|
|
||||||
|
|
||||||
def test_create_calls_generate_random(self):
|
def test_create_calls_generate_random(self):
|
||||||
self.session.generateRandom.return_value = [1, 2, 3, 4, 5, 6, 7,
|
self.session.generateRandom.return_value = [1, 2, 3, 4, 5, 6, 7,
|
||||||
8, 9, 10, 11, 12, 13,
|
8, 9, 10, 11, 12, 13,
|
||||||
@ -171,7 +144,7 @@ class WhenTestingP11CryptoPlugin(unittest.TestCase):
|
|||||||
|
|
||||||
self.p11_mock.Mechanism.assert_called_once()
|
self.p11_mock.Mechanism.assert_called_once()
|
||||||
self.session.encrypt.assert_called_once_with(key,
|
self.session.encrypt.assert_called_once_with(key,
|
||||||
self.plugin._pad(payload),
|
payload,
|
||||||
mech)
|
mech)
|
||||||
self.assertEqual(b'\x01\x02\x03\x04\x05', cyphertext)
|
self.assertEqual(b'\x01\x02\x03\x04\x05', cyphertext)
|
||||||
self.assertEqual('{"iv": "AQIDBAUGBwgJCgsMDQ4PEA=="}',
|
self.assertEqual('{"iv": "AQIDBAUGBwgJCgsMDQ4PEA=="}',
|
||||||
@ -181,7 +154,7 @@ class WhenTestingP11CryptoPlugin(unittest.TestCase):
|
|||||||
key = 'key1'
|
key = 'key1'
|
||||||
ct = MagicMock()
|
ct = MagicMock()
|
||||||
self.session.findObjects.return_value = [key]
|
self.session.findObjects.return_value = [key]
|
||||||
self.session.decrypt.return_value = [100, 101, 102, 103] + [12] * 12
|
self.session.decrypt.return_value = [100, 101, 102, 103]
|
||||||
mech = MagicMock()
|
mech = MagicMock()
|
||||||
self.p11_mock.Mechanism.return_value = mech
|
self.p11_mock.Mechanism.return_value = mech
|
||||||
kek_meta_extended = '{"iv": "AQIDBAUGBwgJCgsMDQ4PEA=="}'
|
kek_meta_extended = '{"iv": "AQIDBAUGBwgJCgsMDQ4PEA=="}'
|
||||||
|
3
tox.ini
3
tox.ini
@ -19,6 +19,9 @@ commands = pep8 barbican --ignore=E711 --count --repeat --show-source --exclude=
|
|||||||
[testenv:py27]
|
[testenv:py27]
|
||||||
commands = nosetests {posargs:--with-xcoverage --all-modules --cover-inclusive --traverse-namespace --with-xunit --cover-package=barbican}
|
commands = nosetests {posargs:--with-xcoverage --all-modules --cover-inclusive --traverse-namespace --with-xunit --cover-package=barbican}
|
||||||
|
|
||||||
|
[testenv:coverage]
|
||||||
|
commands = coverage html {posargs:--include="*barbican*"}
|
||||||
|
|
||||||
[testenv:hacking]
|
[testenv:hacking]
|
||||||
commands = {toxinidir}/tools/hacking.sh
|
commands = {toxinidir}/tools/hacking.sh
|
||||||
|
|
||||||
|
Loading…
x
Reference in New Issue
Block a user