1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
|
From 03eed8cd34cd4fb043c11fc99f6bb0b4fbd5728d Mon Sep 17 00:00:00 2001
From: marios <marios@redhat.com>
Date: Fri, 29 Nov 2013 18:23:54 +0200
Subject: [PATCH] Validate CIDR given as ip-prefix in
security-group-rule-create
There was no validation for the provided ip prefix. This just adds
a simple parse using netaddr and explodes with appropriate message.
Also makes sure ip prefix _is_ cidr (192.168.1.1-->192.168.1.1/32).
Validation occurs at the attribute level (API model) as well as at
the db level, where the ethertype is validated against the ip_prefix
address type.
Unit test cases added - bad prefix, unmasked prefix and incorrect
ethertype. Also adds attribute test cases for the added
convert_ip_prefix_to_cidr method
Closes-Bug: 1255338
Conflicts:
neutron/tests/unit/test_security_groups_rpc.py
neutron/tests/unit/test_extension_security_group.py
Change-Id: I71fb8c887963a122a5bd8cfdda800026c1cd3954
(cherry picked from commit 65aa92b0348b7ab8413f359b00825610cdf66607)
---
neutron/common/exceptions.py | 4 +
neutron/db/securitygroups_db.py | 20 +++++
neutron/extensions/securitygroup.py | 18 ++++-
.../tests/unit/test_extension_security_group.py | 86 ++++++++++++++++++++++
4 files changed, 127 insertions(+), 1 deletion(-)
diff --git a/neutron/common/exceptions.py b/neutron/common/exceptions.py
index 88fa6e4..80a75d1 100644
--- a/neutron/common/exceptions.py
+++ b/neutron/common/exceptions.py
@@ -306,3 +306,7 @@ class NetworkVxlanPortRangeError(object):
class DeviceIDNotOwnedByTenant(Conflict):
message = _("The following device_id %(device_id)s is not owned by your "
"tenant or matches another tenants router.")
+
+
+class InvalidCIDR(BadRequest):
+ message = _("Invalid CIDR %(input)s given as IP prefix")
diff --git a/neutron/db/securitygroups_db.py b/neutron/db/securitygroups_db.py
index 2a7d2ef..8868546 100644
--- a/neutron/db/securitygroups_db.py
+++ b/neutron/db/securitygroups_db.py
@@ -16,6 +16,7 @@
#
# @author: Aaron Rosen, Nicira, Inc
+import netaddr
import sqlalchemy as sa
from sqlalchemy import orm
from sqlalchemy.orm import exc
@@ -331,6 +332,7 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
new_rules.add(rule['security_group_id'])
self._validate_port_range(rule)
+ self._validate_ip_prefix(rule)
if rule['remote_ip_prefix'] and rule['remote_group_id']:
raise ext_sg.SecurityGroupRemoteGroupAndRemoteIpPrefix()
@@ -411,6 +413,24 @@ class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase):
if (i['security_group_rule'] == db_rule):
raise ext_sg.SecurityGroupRuleExists(id=id)
+ def _validate_ip_prefix(self, rule):
+ """Check that a valid cidr was specified as remote_ip_prefix
+
+ No need to check that it is in fact an IP address as this is already
+ validated by attribute validators.
+ Check that rule ethertype is consistent with remote_ip_prefix ip type.
+ Add mask to ip_prefix if absent (192.168.1.10 -> 192.168.1.10/32).
+ """
+ input_prefix = rule['remote_ip_prefix']
+ if input_prefix:
+ addr = netaddr.IPNetwork(input_prefix)
+ # set input_prefix to always include the netmask:
+ rule['remote_ip_prefix'] = str(addr)
+ # check consistency of ethertype with addr version
+ if rule['ethertype'] != "IPv%d" % (addr.version):
+ raise ext_sg.SecurityGroupRuleParameterConflict(
+ ethertype=rule['ethertype'], cidr=input_prefix)
+
def get_security_group_rules(self, context, filters=None, fields=None,
sorts=None, limit=None, marker=None,
page_reverse=False):
diff --git a/neutron/extensions/securitygroup.py b/neutron/extensions/securitygroup.py
index 85d499a..3d10b5a 100644
--- a/neutron/extensions/securitygroup.py
+++ b/neutron/extensions/securitygroup.py
@@ -17,6 +17,7 @@
from abc import ABCMeta
from abc import abstractmethod
+import netaddr
from oslo.config import cfg
@@ -102,6 +103,10 @@ class SecurityGroupRuleExists(qexception.InUse):
message = _("Security group rule already exists. Group id is %(id)s.")
+class SecurityGroupRuleParameterConflict(qexception.InvalidInput):
+ message = _("Conflicting value ethertype %(ethertype)s for CIDR %(cidr)s")
+
+
def convert_protocol(value):
if value is None:
return
@@ -152,6 +157,16 @@ def convert_to_uuid_list_or_none(value_list):
return value_list
+def convert_ip_prefix_to_cidr(ip_prefix):
+ if not ip_prefix:
+ return
+ try:
+ cidr = netaddr.IPNetwork(ip_prefix)
+ return str(cidr)
+ except (TypeError, netaddr.AddrFormatError):
+ raise qexception.InvalidCIDR(input=ip_prefix)
+
+
def _validate_name_not_default(data, valid_values=None):
if data == "default":
raise SecurityGroupDefaultAlreadyExists()
@@ -207,7 +222,8 @@ RESOURCE_ATTRIBUTE_MAP = {
'convert_to': convert_ethertype_to_case_insensitive,
'validate': {'type:values': sg_supported_ethertypes}},
'remote_ip_prefix': {'allow_post': True, 'allow_put': False,
- 'default': None, 'is_visible': True},
+ 'default': None, 'is_visible': True,
+ 'convert_to': convert_ip_prefix_to_cidr},
'tenant_id': {'allow_post': True, 'allow_put': False,
'required_by_policy': True,
'is_visible': True},
diff --git a/neutron/tests/unit/test_extension_security_group.py b/neutron/tests/unit/test_extension_security_group.py
index d53e140..f0b1636 100644
--- a/neutron/tests/unit/test_extension_security_group.py
+++ b/neutron/tests/unit/test_extension_security_group.py
@@ -21,11 +21,13 @@ import webob.exc
from neutron.api.v2 import attributes as attr
from neutron.common import constants as const
+from neutron.common import exceptions as n_exc
from neutron.common.test_lib import test_config
from neutron import context
from neutron.db import db_base_plugin_v2
from neutron.db import securitygroups_db
from neutron.extensions import securitygroup as ext_sg
+from neutron.tests import base
from neutron.tests.unit import test_db_plugin
DB_PLUGIN_KLASS = ('neutron.tests.unit.test_extension_security_group.'
@@ -413,6 +415,70 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self.deserialize(self.fmt, res)
self.assertEqual(res.status_int, webob.exc.HTTPBadRequest.code)
+ def test_create_security_group_rule_invalid_ip_prefix(self):
+ name = 'webservers'
+ description = 'my webservers'
+ for bad_prefix in ['bad_ip', 256, "2001:db8:a::123/129", '172.30./24']:
+ with self.security_group(name, description) as sg:
+ sg_id = sg['security_group']['id']
+ remote_ip_prefix = bad_prefix
+ rule = self._build_security_group_rule(
+ sg_id,
+ 'ingress',
+ const.PROTO_NAME_TCP,
+ '22', '22',
+ remote_ip_prefix)
+ res = self._create_security_group_rule(self.fmt, rule)
+ self.assertEqual(res.status_int, webob.exc.HTTPBadRequest.code)
+
+ def test_create_security_group_rule_invalid_ethertype_for_prefix(self):
+ name = 'webservers'
+ description = 'my webservers'
+ test_addr = {'192.168.1.1/24': 'ipv4', '192.168.1.1/24': 'IPv6',
+ '2001:db8:1234::/48': 'ipv6',
+ '2001:db8:1234::/48': 'IPv4'}
+ for prefix, ether in test_addr.iteritems():
+ with self.security_group(name, description) as sg:
+ sg_id = sg['security_group']['id']
+ ethertype = ether
+ remote_ip_prefix = prefix
+ rule = self._build_security_group_rule(
+ sg_id,
+ 'ingress',
+ const.PROTO_NAME_TCP,
+ '22', '22',
+ remote_ip_prefix,
+ None,
+ None,
+ ethertype)
+ res = self._create_security_group_rule(self.fmt, rule)
+ self.assertEqual(res.status_int, webob.exc.HTTPBadRequest.code)
+
+ def test_create_security_group_rule_with_unmasked_prefix(self):
+ name = 'webservers'
+ description = 'my webservers'
+ addr = {'10.1.2.3': {'mask': '32', 'ethertype': 'IPv4'},
+ 'fe80::2677:3ff:fe7d:4c': {'mask': '128', 'ethertype': 'IPv6'}}
+ for ip in addr:
+ with self.security_group(name, description) as sg:
+ sg_id = sg['security_group']['id']
+ ethertype = addr[ip]['ethertype']
+ remote_ip_prefix = ip
+ rule = self._build_security_group_rule(
+ sg_id,
+ 'ingress',
+ const.PROTO_NAME_TCP,
+ '22', '22',
+ remote_ip_prefix,
+ None,
+ None,
+ ethertype)
+ res = self._create_security_group_rule(self.fmt, rule)
+ self.assertEqual(res.status_int, 201)
+ res_sg = self.deserialize(self.fmt, res)
+ prefix = res_sg['security_group_rule']['remote_ip_prefix']
+ self.assertEqual(prefix, '%s/%s' % (ip, addr[ip]['mask']))
+
def test_create_security_group_rule_tcp_protocol_as_number(self):
name = 'webservers'
description = 'my webservers'
@@ -1348,5 +1414,25 @@ class TestSecurityGroups(SecurityGroupDBTestCase):
self.assertEqual(res.status_int, webob.exc.HTTPBadRequest.code)
+class TestConvertIPPrefixToCIDR(base.BaseTestCase):
+
+ def test_convert_bad_ip_prefix_to_cidr(self):
+ for val in ['bad_ip', 256, "2001:db8:a::123/129"]:
+ self.assertRaises(n_exc.InvalidCIDR,
+ ext_sg.convert_ip_prefix_to_cidr, val)
+ self.assertIsNone(ext_sg.convert_ip_prefix_to_cidr(None))
+
+ def test_convert_ip_prefix_no_netmask_to_cidr(self):
+ addr = {'10.1.2.3': '32', 'fe80::2677:3ff:fe7d:4c': '128'}
+ for k, v in addr.iteritems():
+ self.assertEqual(ext_sg.convert_ip_prefix_to_cidr(k),
+ '%s/%s' % (k, v))
+
+ def test_convert_ip_prefix_with_netmask_to_cidr(self):
+ addresses = ['10.1.0.0/16', '10.1.2.3/32', '2001:db8:1234::/48']
+ for addr in addresses:
+ self.assertEqual(ext_sg.convert_ip_prefix_to_cidr(addr), addr)
+
+
class TestSecurityGroupsXML(TestSecurityGroups):
fmt = 'xml'
--
1.8.5.5
|
GitWeb