Merge "Prevent create port forwarding to port which has binding fip" into stable/rocky

This commit is contained in:
Zuul 2019-06-25 19:35:50 +00:00 committed by Gerrit Code Review
commit 0d37e45256
4 changed files with 102 additions and 2 deletions

View File

@ -33,3 +33,10 @@ class PortHasPortForwarding(n_exc.BadRequest):
class FipInUseByPortForwarding(n_exc.InUse):
message = _("Floating IP %(id)s in use by Port Forwarding resources.")
class PortHasBindingFloatingIP(n_exc.InUse):
message = _("Cannot create port forwarding to floating IP "
"%(floating_ip_address)s (%(fip_id)s) with port %(port_id)s "
"using fixed IP %(fixed_ip)s, as that port already "
"has a binding floating IP.")

View File

@ -301,12 +301,26 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
raise lib_exc.BadRequest(resource=apidef.RESOURCE_NAME,
msg=message)
def _check_port_has_binding_floating_ip(self, context, port_forwarding):
port_id = port_forwarding['internal_port_id']
floatingip_objs = router.FloatingIP.get_objects(
context.elevated(),
fixed_port_id=port_id)
if floatingip_objs:
floating_ip_address = floatingip_objs[0].floating_ip_address
raise pf_exc.PortHasBindingFloatingIP(
floating_ip_address=floating_ip_address,
fip_id=floatingip_objs[0].id,
port_id=port_id,
fixed_ip=port_forwarding['internal_ip_address'])
@db_base_plugin_common.convert_result_to_dict
def create_floatingip_port_forwarding(self, context, floatingip_id,
port_forwarding):
port_forwarding = port_forwarding.get(apidef.RESOURCE_NAME)
port_forwarding['floatingip_id'] = floatingip_id
self._check_port_has_binding_floating_ip(context, port_forwarding)
with db_api.context_manager.writer.using(context):
fip_obj = self._get_fip_obj(context, floatingip_id)
if fip_obj.fixed_port_id:
@ -353,6 +367,8 @@ class PortForwardingPlugin(fip_pf.PortForwardingPluginBase):
new_internal_port_id = None
if port_forwarding and port_forwarding.get('internal_port_id'):
new_internal_port_id = port_forwarding.get('internal_port_id')
self._check_port_has_binding_floating_ip(context, port_forwarding)
try:
with db_api.context_manager.writer.using(context):
fip_obj = self._get_fip_obj(context, floatingip_id)

View File

@ -155,6 +155,46 @@ class PortForwardingTestCase(PortForwardingTestCaseBase):
self.pf_plugin.create_floatingip_port_forwarding,
self.context, self.fip['id'], self.port_forwarding)
def test_create_port_forwarding_port_in_used_by_fip(self):
normal_fip = self._create_floatingip(self.ext_net['id'])
self._update_floatingip(normal_fip['id'], {'port_id': self.port['id']})
self.assertRaises(
pf_exc.PortHasBindingFloatingIP,
self.pf_plugin.create_floatingip_port_forwarding,
self.context, self.fip['id'], self.port_forwarding)
def test_update_port_forwarding_port_in_used_by_fip(self):
normal_fip = self._create_floatingip(self.ext_net['id'])
normal_port = self._create_port(
self.fmt, self.net['id']).json['port']
self._update_floatingip(
normal_fip['id'], {'port_id': normal_port['id']})
res = self.pf_plugin.create_floatingip_port_forwarding(
self.context, self.fip['id'], self.port_forwarding)
expect = {
"external_port": 2225,
"internal_port": 25,
"internal_port_id": self.port['id'],
"protocol": "tcp",
"internal_ip_address": self.port['fixed_ips'][0]['ip_address'],
'id': mock.ANY,
'router_id': self.router['id'],
'floating_ip_address': self.fip['floating_ip_address'],
'floatingip_id': self.fip['id']}
self.assertEqual(expect, res)
# Directly update port forwarding to a port which already has
# bound floating IP.
self.port_forwarding[apidef.RESOURCE_NAME].update(
{apidef.INTERNAL_PORT_ID: normal_port['id'],
apidef.INTERNAL_IP_ADDRESS:
normal_port['fixed_ips'][0]['ip_address']})
self.assertRaises(
pf_exc.PortHasBindingFloatingIP,
self.pf_plugin.update_floatingip_port_forwarding,
self.context, res['id'], self.fip['id'], self.port_forwarding)
def test_update_floatingip_port_forwarding(self):
# create a test port forwarding
res = self.pf_plugin.create_floatingip_port_forwarding(

View File

@ -189,6 +189,8 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
self.pf_plugin.update_floatingip_port_forwarding,
self.ctxt, 'pf_id', **pf_input)
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_check_port_has_binding_floating_ip')
@mock.patch.object(obj_base.NeutronDbObject, 'update_objects')
@mock.patch.object(resources_rpc.ResourcesPushRpcApi, 'push')
@mock.patch.object(pf_plugin.PortForwardingPlugin, '_check_router_match')
@ -199,7 +201,8 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
@mock.patch('neutron.objects.port_forwarding.PortForwarding')
def test_create_floatingip_port_forwarding(
self, mock_port_forwarding, mock_fip_get_object, mock_find_router,
mock_check_router_match, mock_push_api, mock_update_objects):
mock_check_router_match, mock_push_api, mock_update_objects,
mock_check_bind_fip):
# Update fip
pf_input = {
'port_forwarding':
@ -240,6 +243,8 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
mock_push_api.assert_called_once_with(
self.ctxt, mock.ANY, rpc_events.CREATED)
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_check_port_has_binding_floating_ip')
@mock.patch.object(pf_plugin.PortForwardingPlugin,
'_find_existing_port_forwarding')
@mock.patch.object(pf_plugin.PortForwardingPlugin,
@ -252,7 +257,8 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
def test_negative_create_floatingip_port_forwarding(
self, mock_port_forwarding, mock_fip_get_object,
mock_find_router,
mock_check_router_match, mock_try_find_exist):
mock_check_router_match, mock_try_find_exist,
mock_check_bind_fip):
pf_input = {
'port_forwarding': {
'internal_ip_address': '1.1.1.1',
@ -330,3 +336,34 @@ class TestPortForwardingPlugin(testlib_api.SqlTestCase):
self.assertRaises(lib_exc.BadRequest,
self.pf_plugin._check_router_match,
self.ctxt, fip_obj, router_id, pf_dict)
@mock.patch.object(router.FloatingIP, 'get_objects')
def test_create_floatingip_port_forwarding_port_in_use(
self, mock_fip_get_objects):
pf_input = {
'port_forwarding':
{'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'internal_port_id': 'internal_neutron_port',
'floatingip_id': 'fip_id_1'}},
'floatingip_id': 'fip_id_1'}
fip_obj = mock.Mock(floating_ip_address="10.10.10.10")
mock_fip_get_objects.return_value = [fip_obj]
self.assertRaises(pf_exc.PortHasBindingFloatingIP,
self.pf_plugin.create_floatingip_port_forwarding,
self.ctxt, **pf_input)
@mock.patch.object(router.FloatingIP, 'get_objects')
def test_update_floatingip_port_forwarding_port_in_use(
self, mock_fip_get_objects):
pf_input = {
'port_forwarding':
{'port_forwarding': {
'internal_ip_address': '1.1.1.1',
'internal_port_id': 'internal_neutron_port',
'floatingip_id': 'fip_id_2'}}}
fip_obj = mock.Mock(floating_ip_address="10.10.10.11")
mock_fip_get_objects.return_value = [fip_obj]
self.assertRaises(pf_exc.PortHasBindingFloatingIP,
self.pf_plugin.update_floatingip_port_forwarding,
self.ctxt, 'fake-pf-id', 'fip_id_2', **pf_input)