openstack/octavia
Code Issues Proposed changes

Merge "Fix amphorav2 bytes error"

Browse Source
This commit is contained in:
Zuul 2020-11-19 11:24:57 +00:00 committed by Gerrit Code Review
commit 30bbcdd3b2
4 changed files with 37 additions and 17 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
octavia/api/drivers/amphora_driver/v2/driver.py
View File

@ -136,17 +136,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
@ -154,7 +167,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)

4
octavia/api/drivers/utils.py
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

3
octavia/common/data_models.py
View File

@ -43,6 +43,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] = []

24
octavia/tests/unit/api/drivers/amphora_driver/v2/test_amphora_driver.py
View File

@ -744,9 +744,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
@ -757,21 +757,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])