From 83a621e9dc2d890086f0d46052cf7a18b7a9b356 Mon Sep 17 00:00:00 2001 From: Amir Sadoughi Date: Fri, 9 Oct 2015 19:24:45 +0000 Subject: [PATCH] Validate ip_address input on update_port and get_ports JIRA:NCP-1667 --- quark/plugin_modules/ports.py | 14 ++++++++++++-- quark/tests/plugin_modules/test_ports.py | 17 +++++++++++++++++ 2 files changed, 29 insertions(+), 2 deletions(-) diff --git a/quark/plugin_modules/ports.py b/quark/plugin_modules/ports.py index 5140bcc..a77c0ed 100644 --- a/quark/plugin_modules/ports.py +++ b/quark/plugin_modules/ports.py @@ -333,7 +333,12 @@ def update_port(context, id, port): msg="subnet_id required for ip_address allocation") if subnet_id and ip_address: - ip_netaddr = netaddr.IPAddress(ip_address).ipv6() + ip_netaddr = None + try: + ip_netaddr = netaddr.IPAddress(ip_address).ipv6() + except netaddr.AddrFormatError: + raise exceptions.InvalidInput( + error_message="Invalid format provided for ip_address") ip_addresses[ip_netaddr] = subnet_id else: subnet_ids.append(subnet_id) @@ -452,7 +457,12 @@ def get_ports(context, limit=None, sorts=None, marker=None, page_reverse=False, if "ip_address" in filters: if not context.is_admin: raise exceptions.NotAuthorized() - ips = [netaddr.IPAddress(ip) for ip in filters.pop("ip_address")] + ips = [] + try: + ips = [netaddr.IPAddress(ip) for ip in filters.pop("ip_address")] + except netaddr.AddrFormatError: + raise exceptions.InvalidInput( + error_message="Invalid format provided for ip_address") query = db_api.port_find_by_ip_address(context, ip_address=ips, scope=db_api.ALL, **filters) ports = [] diff --git a/quark/tests/plugin_modules/test_ports.py b/quark/tests/plugin_modules/test_ports.py index 5280cf2..365ece3 100644 --- a/quark/tests/plugin_modules/test_ports.py +++ b/quark/tests/plugin_modules/test_ports.py @@ -192,6 +192,13 @@ class TestQuarkGetPortsByIPAddress(test_quark_plugin.TestQuarkPlugin): self.plugin.get_ports(self.context, filters=filters, fields=None) + def test_port_list_malformed_address_bad_request(self): + with self._stubs(ports=[]): + filters = {"ip_address": ["malformed-address-here"]} + admin_ctx = self.context.elevated() + with self.assertRaises(exceptions.BadRequest): + self.plugin.get_ports(admin_ctx, filters=filters, fields=None) + class TestQuarkCreatePortFailure(test_quark_plugin.TestQuarkPlugin): @contextlib.contextmanager @@ -689,6 +696,16 @@ class TestQuarkUpdatePort(test_quark_plugin.TestQuarkPlugin): with self.assertRaises(exceptions.BadRequest): self.plugin.update_port(self.context, 1, new_port) + def test_update_port_fixed_ip_bad_request_malformed_address(self): + with self._stubs( + port=dict(id=1, name="myport", mac_address="0:0:0:0:0:1") + ) as (port_find, port_update, alloc_ip, dealloc_ip): + new_port = dict(port=dict( + fixed_ips=[dict(subnet_id=1, + ip_address="malformed-address-here")])) + with self.assertRaises(exceptions.BadRequest): + self.plugin.update_port(self.context, 1, new_port) + def test_update_port_fixed_ip(self): with self._stubs( port=dict(id=1, name="myport", mac_address="0:0:0:0:0:1")