Fix amphorav2 bytes error

Fix Object of type 'bytes' is not JSON serializable
error while writing barbican tls data to persistence database.
Also fix _encrypt_listener_dict to encrypt dicts properly:
error - data must be bytes (HTTP 500)

Story: 2007877
Task: 40245

Change-Id: If69b5973e2a82c8e8d323a89eefb426a3986cd8f
This commit is contained in:
Ann Taraday 2020-06-29 13:25:49 +04:00
parent 97faa9c2ef
commit 02dc98814b
4 changed files with 37 additions and 17 deletions

View File

@ -133,17 +133,30 @@ class AmphoraProviderDriver(driver_base.ProviderDriver):
consts.LOAD_BALANCER_UPDATES: lb_dict}
self.client.cast({}, 'update_load_balancer', **payload)
def _encrypt_tls_container_data(self, tls_container_data):
for key, val in tls_container_data.items():
if isinstance(val, bytes):
tls_container_data[key] = self.fernet.encrypt(val)
elif isinstance(val, list):
encrypt_vals = []
for i in val:
if isinstance(i, bytes):
encrypt_vals.append(self.fernet.encrypt(i))
else:
encrypt_vals.append(i)
tls_container_data[key] = encrypt_vals
def _encrypt_listener_dict(self, listener_dict):
# We need to encrypt the user cert/key data for sending it
# over messaging.
if listener_dict.get(consts.DEFAULT_TLS_CONTAINER_DATA, False):
listener_dict[consts.DEFAULT_TLS_CONTAINER_DATA] = (
self.fernet.encrypt(
listener_dict[consts.DEFAULT_TLS_CONTAINER_DATA]))
container_data = listener_dict[consts.DEFAULT_TLS_CONTAINER_DATA]
self._encrypt_tls_container_data(container_data)
if listener_dict.get(consts.SNI_CONTAINER_DATA, False):
sni_list = []
for sni_data in listener_dict[consts.SNI_CONTAINER_DATA]:
sni_list.append(self.fernet.encrypt(sni_data))
self._encrypt_tls_container_data(sni_data)
sni_list.append(sni_data)
if sni_list:
listener_dict[consts.SNI_CONTAINER_DATA] = sni_list
@ -151,7 +164,7 @@ class AmphoraProviderDriver(driver_base.ProviderDriver):
def listener_create(self, listener):
self._validate_alpn_protocols(listener)
payload = {consts.LISTENER: listener.to_dict()}
self._encrypt_listener_dict(payload)
self._encrypt_listener_dict(payload[consts.LISTENER])
self.client.cast({}, 'create_listener', **payload)

View File

@ -206,6 +206,10 @@ def _get_secret_data(cert_manager, project_id, secret_ref, for_delete=False):
secret_data = None
else:
raise exceptions.CertificateRetrievalException(ref=secret_ref)
# We need to have json convertible data for storing it in
# persistence jobboard backend.
if isinstance(secret_data, bytes):
return secret_data.decode()
return secret_data

View File

@ -40,6 +40,9 @@ class BaseDataModel(object):
if isinstance(value, datetime.datetime):
ret[attr] = value.isoformat()
continue
if isinstance(value, bytes):
ret[attr] = value.decode()
continue
if recurse:
if isinstance(getattr(self, attr), list):
ret[attr] = []

View File

@ -746,9 +746,9 @@ class TestAmphoraDriver(base.TestRpc):
def test_encrypt_listener_dict(self, mock_fernet):
mock_fern = mock.MagicMock()
mock_fernet.return_value = mock_fern
TEST_DATA = 'some data'
TEST_DATA2 = 'more data'
FAKE_ENCRYPTED_DATA = 'alqwkhjetrhth'
TEST_DATA = {'cert': b'some data'}
TEST_DATA2 = {'test': 'more data'}
FAKE_ENCRYPTED_DATA = b'alqwkhjetrhth'
mock_fern.encrypt.return_value = FAKE_ENCRYPTED_DATA
# We need a class instance with the mock
@ -759,21 +759,21 @@ class TestAmphoraDriver(base.TestRpc):
amp_driver._encrypt_listener_dict(list_dict)
mock_fern.encrypt.assert_called_once_with(TEST_DATA)
mock_fern.encrypt.assert_called_once_with(b'some data')
self.assertEqual(FAKE_ENCRYPTED_DATA,
self.assertEqual({'cert': FAKE_ENCRYPTED_DATA},
list_dict[consts.DEFAULT_TLS_CONTAINER_DATA])
mock_fern.reset_mock()
# Test just sni_container_data
list_dict = {consts.SNI_CONTAINER_DATA: [TEST_DATA, TEST_DATA2]}
TEST_DATA = {'cert': b'some data'}
sni_dict = {consts.SNI_CONTAINER_DATA: [TEST_DATA, TEST_DATA2]}
amp_driver._encrypt_listener_dict(list_dict)
amp_driver._encrypt_listener_dict(sni_dict)
calls = [mock.call(TEST_DATA), mock.call(TEST_DATA2)]
mock_fern.encrypt.assert_called_once_with(b'some data')
mock_fern.encrypt.assert_has_calls(calls)
encrypted_sni = [FAKE_ENCRYPTED_DATA, FAKE_ENCRYPTED_DATA]
self.assertEqual(encrypted_sni, list_dict[consts.SNI_CONTAINER_DATA])
encrypted_sni = [{'cert': FAKE_ENCRYPTED_DATA},
TEST_DATA2]
self.assertEqual(encrypted_sni, sni_dict[consts.SNI_CONTAINER_DATA])