--- /dev/null
+# Copyright (c) 2012 OpenStack, LLC.
+#
+# Licensed under the Apache License, Version 2.0 (the 'License');
+# you may not use this file except in compliance with the License.
+# You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing, software
+# distributed under the License is distributed on an 'AS IS' BASIS,
+# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
+# implied.
+# See the License for the specific language governing permissions and
+# limitations under the License.
+
+ATTR_NOT_SPECIFIED = object()
+
+# Note: a default of ATTR_NOT_SPECIFIED indicates that an
+# attribute is not required, but will be generated by the plugin
+# if it is not specified. Particularly, a value of ATTR_NOT_SPECIFIED
+# is different from an attribute that has been specified with a value of
+# None. For example, if 'gateway_ip' is ommitted in a request to
+# create a subnet, the plugin will receive ATTR_NOT_SPECIFIED
+# and the default gateway_ip will be generated.
+# However, if gateway_ip is specified as None, this means that
+# the subnet does not have a gateway IP.
+
+import logging
+import netaddr
+import re
+
+LOG = logging.getLogger(__name__)
+
+
+def _validate_values(data, valid_values=None):
+ if data in valid_values:
+ return
+ else:
+ msg_dict = dict(data=data, values=valid_values)
+ msg = _("%(data)s is not in %(values)s") % msg_dict
+ LOG.debug("validate_values: %s", msg)
+ return msg
+
+
+def _validate_mac_address(data, valid_values=None):
+ try:
+ netaddr.EUI(data)
+ return
+ except Exception:
+ msg = _("%s is not a valid MAC address") % data
+ LOG.debug("validate_mac_address: %s", msg)
+ return msg
+
+
+def _validate_ip_address(data, valid_values=None):
+ try:
+ netaddr.IPAddress(data)
+ return
+ except Exception:
+ msg = _("%s is not a valid IP address") % data
+ LOG.debug("validate_ip_address: %s", msg)
+ return msg
+
+
+def _validate_subnet(data, valid_values=None):
+ try:
+ netaddr.IPNetwork(data)
+ return
+ except Exception:
+ msg = _("%s is not a valid IP subnet") % data
+ LOG.debug("validate_subnet: %s", msg)
+ return msg
+
+
+def _validate_regex(data, valid_values=None):
+ match = re.match(valid_values, data)
+ if match:
+ return
+ else:
+ msg = _("%s is not valid") % data
+ LOG.debug("validate_regex: %s", msg)
+ return msg
+
+
+HEX_ELEM = '[0-9A-Fa-f]'
+UUID_PATTERN = '-'.join([HEX_ELEM + '{8}', HEX_ELEM + '{4}',
+ HEX_ELEM + '{4}', HEX_ELEM + '{4}',
+ HEX_ELEM + '{12}'])
+
+# Dictionary that maintains a list of validation functions
+validators = {'type:values': _validate_values,
+ 'type:mac_address': _validate_mac_address,
+ 'type:ip_address': _validate_ip_address,
+ 'type:subnet': _validate_subnet,
+ 'type:regex': _validate_regex}
+
+# Note: a default of ATTR_NOT_SPECIFIED indicates that an
+# attribute is not required, but will be generated by the plugin
+# if it is not specified. Particularly, a value of ATTR_NOT_SPECIFIED
+# is different from an attribute that has been specified with a value of
+# None. For example, if 'gateway_ip' is ommitted in a request to
+# create a subnet, the plugin will receive ATTR_NOT_SPECIFIED
+# and the default gateway_ip will be generated.
+# However, if gateway_ip is specified as None, this means that
+# the subnet does not have a gateway IP.
+# Some of the following attributes are used by the policy engine.
+# They are explicitly marked with the required_by_policy flag to ensure
+# they are always returned by a plugin for policy processing, even if
+# they are not specified in the 'fields' query param
+RESOURCE_ATTRIBUTE_MAP = {
+ 'networks': {
+ 'id': {'allow_post': False, 'allow_put': False,
+ 'validate': {'type:regex': UUID_PATTERN}},
+ 'name': {'allow_post': True, 'allow_put': True},
+ 'subnets': {'allow_post': True, 'allow_put': True, 'default': []},
+ 'admin_state_up': {'allow_post': True, 'allow_put': True,
+ 'default': True,
+ 'validate': {'type:values': [True, False]}},
+ 'status': {'allow_post': False, 'allow_put': False},
+ 'tenant_id': {'allow_post': True, 'allow_put': False,
+ 'required_by_policy': True},
+ },
+ 'ports': {
+ 'id': {'allow_post': False, 'allow_put': False,
+ 'validate': {'type:regex': UUID_PATTERN}},
+ 'network_id': {'allow_post': True, 'allow_put': False,
+ 'validate': {'type:regex': UUID_PATTERN}},
+ 'admin_state_up': {'allow_post': True, 'allow_put': True,
+ 'default': True,
+ 'validate': {'type:values': [True, False]}},
+ 'mac_address': {'allow_post': True, 'allow_put': False,
+ 'default': ATTR_NOT_SPECIFIED,
+ 'validate': {'type:mac_address': None}},
+ 'fixed_ips': {'allow_post': True, 'allow_put': True,
+ 'default': ATTR_NOT_SPECIFIED},
+ 'host_routes': {'allow_post': True, 'allow_put': True,
+ 'default': ATTR_NOT_SPECIFIED},
+ 'device_id': {'allow_post': True, 'allow_put': True, 'default': ''},
+ 'tenant_id': {'allow_post': True, 'allow_put': False,
+ 'required_by_policy': True},
+ },
+ 'subnets': {
+ 'id': {'allow_post': False, 'allow_put': False,
+ 'validate': {'type:regex': UUID_PATTERN}},
+ 'ip_version': {'allow_post': True, 'allow_put': False,
+ 'validate': {'type:values': [4, 6]}},
+ 'network_id': {'allow_post': True, 'allow_put': False,
+ 'validate': {'type:regex': UUID_PATTERN}},
+ 'cidr': {'allow_post': True, 'allow_put': False,
+ 'validate': {'type:subnet': None}},
+ 'gateway_ip': {'allow_post': True, 'allow_put': True,
+ 'default': ATTR_NOT_SPECIFIED,
+ 'validate': {'type:ip_address': None}},
+ #TODO(salvatore-orlando): Enable PUT on allocation_pools
+ 'allocation_pools': {'allow_post': True, 'allow_put': False,
+ 'default': ATTR_NOT_SPECIFIED},
+ 'dns_namesevers': {'allow_post': True, 'allow_put': True,
+ 'default': ATTR_NOT_SPECIFIED},
+ 'additional_host_routes': {'allow_post': True, 'allow_put': True,
+ 'default': ATTR_NOT_SPECIFIED},
+ 'tenant_id': {'allow_post': True, 'allow_put': False,
+ 'required_by_policy': True},
+ }
+}
# limitations under the License.
import logging
-
import webob.exc
+from quantum.api.v2 import attributes
from quantum.api.v2 import resource as wsgi_resource
from quantum.api.v2 import views
from quantum.common import exceptions
msg = _("Cannot update read-only attribute %s") % attr
raise webob.exc.HTTPUnprocessableEntity(msg)
+ # Check that configured values are correct
+ for attr, attr_vals in self._attr_info.iteritems():
+ if not ('validate' in attr_vals and
+ attr in res_dict and
+ res_dict[attr] != attributes.ATTR_NOT_SPECIFIED):
+ continue
+ for rule in attr_vals['validate']:
+ res = attributes.validators[rule](res_dict[attr],
+ attr_vals['validate'][rule])
+ if res:
+ msg_dict = dict(attr=attr, reason=res)
+ msg = _("Invalid input for %(attr)s. "
+ "Reason: %(reason)s.") % msg_dict
+ raise webob.exc.HTTPUnprocessableEntity(msg)
+
return body
def _validate_network_tenant_ownership(self, request, resource_item):
import webob.dec
import webob.exc
+from quantum.api.v2 import attributes
from quantum.api.v2 import base
from quantum import manager
from quantum.openstack.common import cfg
LOG = logging.getLogger(__name__)
-HEX_ELEM = '[0-9A-Fa-f]'
-UUID_PATTERN = '-'.join([HEX_ELEM + '{8}', HEX_ELEM + '{4}',
- HEX_ELEM + '{4}', HEX_ELEM + '{4}',
- HEX_ELEM + '{12}'])
COLLECTION_ACTIONS = ['index', 'create']
MEMBER_ACTIONS = ['show', 'update', 'delete']
-REQUIREMENTS = {'id': UUID_PATTERN, 'format': 'xml|json'}
-
-
-ATTR_NOT_SPECIFIED = object()
-
-# Note: a default of ATTR_NOT_SPECIFIED indicates that an
-# attribute is not required, but will be generated by the plugin
-# if it is not specified. Particularly, a value of ATTR_NOT_SPECIFIED
-# is different from an attribute that has been specified with a value of
-# None. For example, if 'gateway_ip' is ommitted in a request to
-# create a subnet, the plugin will receive ATTR_NOT_SPECIFIED
-# and the default gateway_ip will be generated.
-# However, if gateway_ip is specified as None, this means that
-# the subnet does not have a gateway IP.
-# Some of the following attributes are used by the policy engine.
-# They are explicitly marked with the required_by_policy flag to ensure
-# they are always returned by a plugin for policy processing, even if
-# they are not specified in the 'fields' query param
-
-RESOURCE_ATTRIBUTE_MAP = {
- 'networks': {
- 'id': {'allow_post': False, 'allow_put': False},
- 'name': {'allow_post': True, 'allow_put': True},
- 'subnets': {'allow_post': True, 'allow_put': True, 'default': []},
- 'admin_state_up': {'allow_post': True, 'allow_put': True,
- 'default': True},
- 'status': {'allow_post': False, 'allow_put': False},
- 'tenant_id': {'allow_post': True, 'allow_put': False,
- 'required_by_policy': True},
- },
- 'ports': {
- 'id': {'allow_post': False, 'allow_put': False},
- 'network_id': {'allow_post': True, 'allow_put': False},
- 'admin_state_up': {'allow_post': True, 'allow_put': True,
- 'default': True},
- 'mac_address': {'allow_post': True, 'allow_put': False,
- 'default': ATTR_NOT_SPECIFIED},
- 'fixed_ips': {'allow_post': True, 'allow_put': True,
- 'default': ATTR_NOT_SPECIFIED},
- 'host_routes': {'allow_post': True, 'allow_put': True,
- 'default': ATTR_NOT_SPECIFIED},
- 'device_id': {'allow_post': True, 'allow_put': True, 'default': ''},
- 'tenant_id': {'allow_post': True, 'allow_put': False,
- 'required_by_policy': True},
- },
- 'subnets': {
- 'id': {'allow_post': False, 'allow_put': False},
- 'ip_version': {'allow_post': True, 'allow_put': False},
- 'network_id': {'allow_post': True, 'allow_put': False},
- 'cidr': {'allow_post': True, 'allow_put': False},
- 'gateway_ip': {'allow_post': True, 'allow_put': True,
- 'default': ATTR_NOT_SPECIFIED},
- #TODO(salvatore-orlando): Enable PUT on allocation_pools
- 'allocation_pools': {'allow_post': True, 'allow_put': False,
- 'default': ATTR_NOT_SPECIFIED},
- 'dns_namesevers': {'allow_post': True, 'allow_put': True,
- 'default': ATTR_NOT_SPECIFIED},
- 'additional_host_routes': {'allow_post': True, 'allow_put': True,
- 'default': ATTR_NOT_SPECIFIED},
- 'tenant_id': {'allow_post': True, 'allow_put': False,
- 'required_by_policy': True},
- }
-}
+REQUIREMENTS = {'id': attributes.UUID_PATTERN, 'format': 'xml|json'}
class Index(wsgi.Application):
mapper.connect('index', '/', controller=Index(resources))
for resource in resources:
_map_resource(resources[resource], resource,
- RESOURCE_ATTRIBUTE_MAP.get(resources[resource],
- dict()))
+ attributes.RESOURCE_ATTRIBUTE_MAP.get(
+ resources[resource], dict()))
super(APIRouter, self).__init__(mapper)
from sqlalchemy import orm
from sqlalchemy.orm import exc
-from quantum.api.v2 import router as api_router
+from quantum.api.v2 import attributes
from quantum.common import exceptions as q_exc
from quantum.db import api as db
from quantum.db import models_v2
p = port['port']
ips = []
- fixed_configured = (p['fixed_ips'] != api_router.ATTR_NOT_SPECIFIED)
+ fixed_configured = (p['fixed_ips'] != attributes.ATTR_NOT_SPECIFIED)
if fixed_configured:
configured_ips = self._test_fixed_ips_for_port(context,
p["network_id"],
"""
pools = []
- if subnet['allocation_pools'] == api_router.ATTR_NOT_SPECIFIED:
+ if subnet['allocation_pools'] == attributes.ATTR_NOT_SPECIFIED:
# Auto allocate the pool around gateway
gw_ip = int(netaddr.IPAddress(subnet['gateway_ip']))
net = netaddr.IPNetwork(subnet['cidr'])
s = subnet['subnet']
net = netaddr.IPNetwork(s['cidr'])
- if s['gateway_ip'] == api_router.ATTR_NOT_SPECIFIED:
+ if s['gateway_ip'] == attributes.ATTR_NOT_SPECIFIED:
s['gateway_ip'] = str(netaddr.IPAddress(net.first + 1))
tenant_id = self._get_tenant_id_for_create(context, s)
# Ensure that a MAC address is defined and it is unique on the
# network
- if p['mac_address'] == api_router.ATTR_NOT_SPECIFIED:
+ if p['mac_address'] == attributes.ATTR_NOT_SPECIFIED:
p['mac_address'] = QuantumDbPluginV2._generate_mac(
context, p["network_id"])
else:
from webob import exc
+from quantum.api.v2 import attributes
from quantum.api.v2 import resource as wsgi_resource
from quantum.api.v2 import router
from quantum.api.v2 import views
'device_id': device_id,
'admin_state_up': True}}
full_input = {'port': {'admin_state_up': True,
- 'mac_address': router.ATTR_NOT_SPECIFIED,
- 'fixed_ips': router.ATTR_NOT_SPECIFIED,
- 'host_routes': router.ATTR_NOT_SPECIFIED}}
+ 'mac_address': attributes.ATTR_NOT_SPECIFIED,
+ 'fixed_ips': attributes.ATTR_NOT_SPECIFIED,
+ 'host_routes': attributes.ATTR_NOT_SPECIFIED}}
full_input['port'].update(initial_input['port'])
return_value = {'id': _uuid(), 'status': 'ACTIVE',
'admin_state_up': True,
self.assertEquals(ips[0]['ip_address'], '10.0.1.3')
self.assertEquals(ips[0]['subnet_id'], subnet['subnet']['id'])
+ def test_invalid_admin_state(self):
+ with self.network() as network:
+ data = {'port': {'network_id': network['network']['id'],
+ 'tenant_id': network['network']['tenant_id'],
+ 'admin_state_up': 7,
+ 'fixed_ips': []}}
+ port_req = self.new_create_request('ports', data)
+ res = port_req.get_response(self.api)
+ self.assertEquals(res.status_int, 422)
+
+ def test_invalid_mac_address(self):
+ with self.network() as network:
+ data = {'port': {'network_id': network['network']['id'],
+ 'tenant_id': network['network']['tenant_id'],
+ 'admin_state_up': 1,
+ 'mac_address': 'mac',
+ 'fixed_ips': []}}
+ port_req = self.new_create_request('ports', data)
+ res = port_req.get_response(self.api)
+ self.assertEquals(res.status_int, 422)
+
class TestNetworksV2(QuantumDbPluginV2TestCase):
# NOTE(cerberus): successful network update and delete are
self.assertEquals(res['network']['name'],
net['network']['name'])
+ def test_invalid_admin_status(self):
+ data = {'network': {'name': 'net',
+ 'admin_state_up': 7,
+ 'tenant_id': self._tenant_id}}
+ network_req = self.new_create_request('networks', data)
+ res = network_req.get_response(self.api)
+ self.assertEquals(res.status_int, 422)
+
class TestSubnetsV2(QuantumDbPluginV2TestCase):
subnet['subnet']['cidr'])
self.assertEquals(res2['cidr'],
subnet2['subnet']['cidr'])
+
+ def test_invalid_ip_version(self):
+ with self.network() as network:
+ data = {'subnet': {'network_id': network['network']['id'],
+ 'cidr': '10.0.2.0/24',
+ 'ip_version': 7,
+ 'tenant_id': network['network']['tenant_id'],
+ 'gateway_ip': '10.0.2.1'}}
+
+ subnet_req = self.new_create_request('subnets', data)
+ res = subnet_req.get_response(self.api)
+ self.assertEquals(res.status_int, 422)
+
+ def test_invalid_subnet(self):
+ with self.network() as network:
+ data = {'subnet': {'network_id': network['network']['id'],
+ 'cidr': 'invalid',
+ 'ip_version': 4,
+ 'tenant_id': network['network']['tenant_id'],
+ 'gateway_ip': '10.0.2.1'}}
+
+ subnet_req = self.new_create_request('subnets', data)
+ res = subnet_req.get_response(self.api)
+ self.assertEquals(res.status_int, 422)
+
+ def test_invalid_ip_address(self):
+ with self.network() as network:
+ data = {'subnet': {'network_id': network['network']['id'],
+ 'cidr': '10.0.2.0/24',
+ 'ip_version': 4,
+ 'tenant_id': network['network']['tenant_id'],
+ 'gateway_ip': 'ipaddress'}}
+
+ subnet_req = self.new_create_request('subnets', data)
+ res = subnet_req.get_response(self.api)
+ self.assertEquals(res.status_int, 422)
+
+ def test_invalid_uuid(self):
+ with self.network() as network:
+ data = {'subnet': {'network_id': 'invalid-uuid',
+ 'cidr': '10.0.2.0/24',
+ 'ip_version': 4,
+ 'tenant_id': network['network']['tenant_id'],
+ 'gateway_ip': '10.0.0.1'}}
+
+ subnet_req = self.new_create_request('subnets', data)
+ res = subnet_req.get_response(self.api)
+ self.assertEquals(res.status_int, 422)
Code Review / openstack-build / neutron-build.git / commitdiff