openstack/neutron
Code Issues Proposed changes

Catch invalid subnet service_types on input

Browse Source 
_validate_subnet_service_types function does not consider the
data type of service_type on creatation or update. For example,
if an integer type is passed, the neutron-server will hit
internal error.

This patch adds a check to make sure the data type is unicode.

Change-Id: I5e6d15def3e23f51172b69e1287ff18ab5d3f6aa
Closes-Bug: #1633066
This commit is contained in:
ZhaoBo 2016-10-14 00:52:27 +08:00
parent 1111fc61a1
commit a594df67a7
2 changed files with 17 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
neutron/extensions/subnet_service_types.py
View File

@ -14,6 +14,7 @@ from neutron_lib.api import extensions
from neutron_lib.api import validators
from neutron_lib import constants
from neutron_lib import exceptions
import six
import webob.exc
from neutron._i18n import _
@ -29,6 +30,10 @@ class InvalidSubnetServiceType(exceptions.InvalidInput):
"to a valid device owner.")
class InvalidInputSubnetServiceType(exceptions.InvalidInput):
message = _("Subnet service type %(service_type)s is not a string.")
def _validate_subnet_service_types(service_types, valid_values=None):
if service_types:
if not isinstance(service_types, list):
@ -41,7 +46,9 @@ def _validate_subnet_service_types(service_types, valid_values=None):
prefixes += constants.DEVICE_OWNER_COMPUTE_PREFIX
for service_type in service_types:
if not service_type.startswith(tuple(prefixes)):
if not isinstance(service_type, six.text_type):
raise InvalidInputSubnetServiceType(service_type=service_type)
elif not service_type.startswith(tuple(prefixes)):
raise InvalidSubnetServiceType(service_type=service_type)

15
neutron/tests/unit/extensions/test_subnet_service_types.py
View File

@ -98,6 +98,7 @@ class SubnetServiceTypesExtensionTestCase(
def test_create_subnet_invalid_type(self):
self._test_create_subnet(['foo'], expect_fail=True)
self._test_create_subnet([1], expect_fail=True)
def test_create_subnet_no_type(self):
res = self._create_service_subnet()
@ -105,12 +106,12 @@ class SubnetServiceTypesExtensionTestCase(
subnet = subnet['subnet']
self.assertFalse(subnet['service_types'])
def _test_update_subnet(self, subnet, service_types, expect_fail=False):
def _test_update_subnet(self, subnet, service_types, fail_code=None):
data = {'subnet': {'service_types': service_types}}
req = self.new_update_request('subnets', data, subnet['id'])
res = self.deserialize(self.fmt, req.get_response(self.api))
if expect_fail:
self.assertEqual('InvalidSubnetServiceType',
if fail_code is not None:
self.assertEqual(fail_code,
res['NeutronError']['type'])
else:
subnet = res['subnet']
@ -155,12 +156,14 @@ class SubnetServiceTypesExtensionTestCase(
self._test_update_subnet(subnet, service_types)
def test_update_subnet_invalid_type(self):
service_types = ['foo']
# Create a subnet with no service type
res = self._create_service_subnet()
subnet = self.deserialize('json', res)['subnet']
# Update it with an invalid service type
self._test_update_subnet(subnet, service_types, expect_fail=True)
# Update it with invalid service type(s)
self._test_update_subnet(subnet, ['foo'],
fail_code='InvalidSubnetServiceType')
self._test_update_subnet(subnet, [2],
fail_code='InvalidInputSubnetServiceType')
def _assert_port_res(self, port, service_type, subnet, fallback,
error='IpAddressGenerationFailureNoMatchingSubnet'):