openstack/octavia
Code Issues Proposed changes

Merge "Deny the creation of L7Policies for HTTPS/TCP/UDP listeners" into stable/yoga

Browse Source
This commit is contained in:
Zuul 2022-05-03 17:06:37 +00:00 committed by Gerrit Code Review
commit c6da46ea8d
5 changed files with 71 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

13
octavia/api/drivers/amphora_driver/v1/driver.py
View File

@ -56,6 +56,11 @@ AMPHORA_SUPPORTED_PROTOCOLS = [
lib_consts.PROTOCOL_PROMETHEUS,
]
VALID_L7POLICY_LISTENER_PROTOCOLS = [
lib_consts.PROTOCOL_HTTP,
lib_consts.PROTOCOL_TERMINATED_HTTPS
]
class AmphoraProviderDriver(driver_base.ProviderDriver):
def __init__(self):
@ -340,6 +345,14 @@ class AmphoraProviderDriver(driver_base.ProviderDriver):
# L7 Policy
def l7policy_create(self, l7policy):
db_listener = self.repositories.listener.get(db_apis.get_session(),
id=l7policy.listener_id)
if db_listener.protocol not in VALID_L7POLICY_LISTENER_PROTOCOLS:
msg = ('%s protocol listeners do not support L7 policies' % (
db_listener.protocol))
raise exceptions.UnsupportedOptionError(
user_fault_string=msg,
operator_fault_string=msg)
payload = {consts.L7POLICY_ID: l7policy.l7policy_id}
self.client.cast({}, 'create_l7policy', **payload)

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

@ -57,6 +57,11 @@ AMPHORA_SUPPORTED_PROTOCOLS = [
lib_consts.PROTOCOL_PROMETHEUS,
]
VALID_L7POLICY_LISTENER_PROTOCOLS = [
lib_consts.PROTOCOL_HTTP,
lib_consts.PROTOCOL_TERMINATED_HTTPS
]
class AmphoraProviderDriver(driver_base.ProviderDriver):
def __init__(self):
@ -363,6 +368,14 @@ class AmphoraProviderDriver(driver_base.ProviderDriver):
# L7 Policy
def l7policy_create(self, l7policy):
db_listener = self.repositories.listener.get(db_apis.get_session(),
id=l7policy.listener_id)
if db_listener.protocol not in VALID_L7POLICY_LISTENER_PROTOCOLS:
msg = ('%s protocol listeners do not support L7 policies' % (
db_listener.protocol))
raise exceptions.UnsupportedOptionError(
user_fault_string=msg,
operator_fault_string=msg)
payload = {consts.L7POLICY: l7policy.to_dict()}
self.client.cast({}, 'create_l7policy', **payload)

20
octavia/tests/unit/api/drivers/amphora_driver/v1/test_driver.py
View File

@ -619,14 +619,32 @@ class TestAmphoraDriver(base.TestRpc):
mock_cast.assert_called_with({}, 'update_health_monitor', **payload)
# L7 Policy
@mock.patch('octavia.db.api.get_session')
@mock.patch('octavia.db.repositories.ListenerRepository.get')
@mock.patch('oslo_messaging.RPCClient.cast')
def test_l7policy_create(self, mock_cast):
def test_l7policy_create(self, mock_cast, mock_listener_get, mock_session):
mock_listener = mock.MagicMock()
mock_listener.protocol = consts.PROTOCOL_HTTP
mock_listener_get.return_value = mock_listener
provider_l7policy = driver_dm.L7Policy(
l7policy_id=self.sample_data.l7policy1_id)
self.amp_driver.l7policy_create(provider_l7policy)
payload = {consts.L7POLICY_ID: self.sample_data.l7policy1_id}
mock_cast.assert_called_with({}, 'create_l7policy', **payload)
@mock.patch('octavia.db.api.get_session')
@mock.patch('octavia.db.repositories.ListenerRepository.get')
def test_l7policy_create_invalid_listener_protocol(self, mock_listener_get,
mock_session):
mock_listener = mock.MagicMock()
mock_listener.protocol = consts.PROTOCOL_UDP
mock_listener_get.return_value = mock_listener
provider_l7policy = driver_dm.L7Policy(
l7policy_id=self.sample_data.l7policy1_id)
self.assertRaises(exceptions.UnsupportedOptionError,
self.amp_driver.l7policy_create,
provider_l7policy)
@mock.patch('oslo_messaging.RPCClient.cast')
def test_l7policy_delete(self, mock_cast):
provider_l7policy = driver_dm.L7Policy(

20
octavia/tests/unit/api/drivers/amphora_driver/v2/test_driver.py
View File

@ -674,14 +674,32 @@ class TestAmphoraDriver(base.TestRpc):
mock_cast.assert_called_with({}, 'update_health_monitor', **payload)
# L7 Policy
@mock.patch('octavia.db.api.get_session')
@mock.patch('octavia.db.repositories.ListenerRepository.get')
@mock.patch('oslo_messaging.RPCClient.cast')
def test_l7policy_create(self, mock_cast):
def test_l7policy_create(self, mock_cast, mock_listener_get, mock_session):
mock_listener = mock.MagicMock()
mock_listener.protocol = consts.PROTOCOL_HTTP
mock_listener_get.return_value = mock_listener
provider_l7policy = driver_dm.L7Policy(
l7policy_id=self.sample_data.l7policy1_id)
self.amp_driver.l7policy_create(provider_l7policy)
payload = {consts.L7POLICY: provider_l7policy.to_dict()}
mock_cast.assert_called_with({}, 'create_l7policy', **payload)
@mock.patch('octavia.db.api.get_session')
@mock.patch('octavia.db.repositories.ListenerRepository.get')
def test_l7policy_create_invalid_listener_protocol(self, mock_listener_get,
mock_session):
mock_listener = mock.MagicMock()
mock_listener.protocol = consts.PROTOCOL_UDP
mock_listener_get.return_value = mock_listener
provider_l7policy = driver_dm.L7Policy(
l7policy_id=self.sample_data.l7policy1_id)
self.assertRaises(exceptions.UnsupportedOptionError,
self.amp_driver.l7policy_create,
provider_l7policy)
@mock.patch('oslo_messaging.RPCClient.cast')
def test_l7policy_delete(self, mock_cast):
provider_l7policy = driver_dm.L7Policy(

7
releasenotes/notes/validate-protocols-for-l7policies-83d678171f13136a.yaml Normal file
View File

@ -0,0 +1,7 @@
---
fixes:
- |
Validate that the creation of L7 policies is compatible with the protocol
of the listener in the Amphora driver. L7 policies are allowed for
Terminated HTTPS or HTTP protocol listeners, but not for HTTPS, TCP or UDP
protocols listeners.