openstack/neutron
Code Issues Proposed changes

Validate type of allowed_address_pairs

Browse Source 
If user don't pass "type=dict list=true" in create/update port
with allowed_address_pairs attribute, we will hit Quota exceed
exception. So we need to check address_pairs type in validation,
if allowed address pairs is not Python list, we should raise an
exception.

Change-Id: I7c84b728e8e660b678f251122cc332cd0ce5c576
Closes-Bug: #1631432
This commit is contained in:
Dongcan Ye 2016-10-18 17:44:04 +08:00
parent d2a28a8c73
commit 9204b34afc
2 changed files with 22 additions and 5 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
neutron/extensions/allowedaddresspairs.py
View File

@ -48,13 +48,12 @@ class AllowedAddressPairExhausted(nexception.BadRequest):
def _validate_allowed_address_pairs(address_pairs, valid_values=None):
unique_check = {}
try:
if len(address_pairs) > cfg.CONF.max_allowed_address_pair:
raise AllowedAddressPairExhausted(
quota=cfg.CONF.max_allowed_address_pair)
except TypeError:
if not isinstance(address_pairs, list):
raise webob.exc.HTTPBadRequest(
_("Allowed address pairs must be a list."))
if len(address_pairs) > cfg.CONF.max_allowed_address_pair:
raise AllowedAddressPairExhausted(
quota=cfg.CONF.max_allowed_address_pair)
for address_pair in address_pairs:
# mac_address is optional, if not set we use the mac on the port

18
neutron/tests/unit/db/test_allowedaddresspairs_db.py
View File

@ -230,6 +230,11 @@ class TestAllowedAddressPairs(AllowedAddressPairDBTestCase):
'icbb': 'agreed'}]
self._create_port_with_address_pairs(address_pairs, 400)
def test_create_port_with_unexpected_address_pairs_format(self):
address_pairs = {'mac_address': '00:00:00:00:00:01',
'ip_address': '10.0.0.1'}
self._create_port_with_address_pairs(address_pairs, 400)
def _create_port_with_address_pairs(self, address_pairs, ret_code):
with self.network() as net:
res = self._create_port(self.fmt, net['network']['id'],
@ -255,6 +260,19 @@ class TestAllowedAddressPairs(AllowedAddressPairDBTestCase):
address_pairs)
self._delete('ports', port['port']['id'])
def test_update_add_address_pairs_with_unexpected_format(self):
with self.network() as net:
res = self._create_port(self.fmt, net['network']['id'])
port = self.deserialize(self.fmt, res)
address_pairs = {'mac_address': '00:00:00:00:00:01',
'ip_address': '10.0.0.1'}
update_port = {'port': {addr_pair.ADDRESS_PAIRS:
address_pairs}}
req = self.new_update_request('ports', update_port,
port['port']['id'])
res = req.get_response(self.api)
self.assertEqual(400, res.status_int)
def test_create_address_gets_port_mac(self):
with self.network() as net:
address_pairs = [{'ip_address': '23.23.23.23'}]