Tweak security port validation for ICMP

Horizon allows for ICMP to be type:code.
Type and code can be from -1 to 255.

API refers to both EC2 and Nova APIs

This patch attempts to resolve:
1. API code throws exceptations when 0 is passed for either field
2. API code validates type:code like from->to range.  type and code
   are independent
3. Update unit tests for this new set of operations.

A side effect is that the following are allowed type:code.
-1:X
X:-1

The code assumes that -1 is a wildcard for the field.

bug 956967

Change-Id: Ieb6989815afc6986b72e0efc7611c2cc353ab5d8

Authors

@@ -68,6 +68,7 @@ François Charlier <francois.charlier@enovance.com>
 Gabe Westmaas <gabe.westmaas@rackspace.com>
 Gary Kotton <garyk@radware.com>
 Gaurav Gupta <gaurav@denali-systems.com>
+Greg Althaus <galthaus@austin.rr.com>
 Hengqing Hu <hudayou@hotmail.com>
 Hisaharu Ishii <ishii.hisaharu@lab.ntt.co.jp>
 Hisaki Ohara <hisaki.ohara@intel.com>

nova/api/ec2/cloud.py

@@ -463,7 +463,7 @@ class CloudController(object):
                 else:
                     for protocol, min_port, max_port in (('icmp', -1, -1),
                                                          ('tcp', 1, 65535),
-                                                         ('udp', 1, 65536)):
+                                                         ('udp', 1, 65535)):
                         r['ipProtocol'] = protocol
                         r['fromPort'] = min_port
                         r['toPort'] = max_port
@@ -559,15 +559,16 @@ class CloudController(object):
             # Open everything if an explicit port range or type/code are not
             # specified, but only if a source group was specified.
             ip_proto_upper = ip_protocol.upper() if ip_protocol else ''
-            if ip_proto_upper == 'ICMP' and not from_port and not to_port:
+            if (ip_proto_upper == 'ICMP' and
+                from_port is None and to_port is None):
                 from_port = -1
                 to_port = -1
-            elif (ip_proto_upper in ['TCP', 'UDP'] and not from_port
+            elif (ip_proto_upper in ['TCP', 'UDP'] and from_port is None
-                  and not to_port):
+                  and to_port is None):
                 from_port = 1
                 to_port = 65535
 
-        if ip_protocol and from_port and to_port:
+        if ip_protocol and from_port is not None and to_port is not None:
 
             ip_protocol = str(ip_protocol)
             try:
@@ -587,7 +588,8 @@ class CloudController(object):
 
             # Verify that from_port must always be less than
             # or equal to to_port
-            if from_port > to_port:
+            if (ip_protocol.upper() in ['TCP', 'UDP'] and
+                (from_port > to_port)):
                 raise exception.InvalidPortRange(from_port=from_port,
                       to_port=to_port, msg="Former value cannot"
                                             " be greater than the later")
@@ -601,7 +603,8 @@ class CloudController(object):
 
             # Verify ICMP type and code
             if (ip_protocol.upper() == "ICMP" and
-                (from_port < -1 or to_port > 255)):
+                (from_port < -1 or from_port > 255 or
+                to_port < -1 or to_port > 255)):
                 raise exception.InvalidPortRange(from_port=from_port,
                       to_port=to_port, msg="For ICMP, the"
                                            " type:code must be valid")

nova/api/openstack/compute/contrib/security_groups.py

@@ -440,15 +440,16 @@ class SecurityGroupRulesController(SecurityGroupControllerBase):
             # Open everything if an explicit port range or type/code are not
             # specified, but only if a source group was specified.
             ip_proto_upper = ip_protocol.upper() if ip_protocol else ''
-            if ip_proto_upper == 'ICMP' and not from_port and not to_port:
+            if (ip_proto_upper == 'ICMP' and
+                from_port is None and to_port is None):
                 from_port = -1
                 to_port = -1
-            elif (ip_proto_upper in ['TCP', 'UDP'] and not from_port
+            elif (ip_proto_upper in ['TCP', 'UDP'] and from_port is None
-                  and not to_port):
+                  and to_port is None):
                 from_port = 1
                 to_port = 65535
 
-        if ip_protocol and from_port and to_port:
+        if ip_protocol and from_port is not None and to_port is not None:
 
             ip_protocol = str(ip_protocol)
             try:
@@ -467,7 +468,8 @@ class SecurityGroupRulesController(SecurityGroupControllerBase):
 
             # Verify that from_port must always be less than
             # or equal to to_port
-            if from_port > to_port:
+            if (ip_protocol.upper() in ['TCP', 'UDP'] and
+                from_port > to_port):
                 raise exception.InvalidPortRange(from_port=from_port,
                       to_port=to_port, msg="Former value cannot"
                                             " be greater than the later")
@@ -481,7 +483,8 @@ class SecurityGroupRulesController(SecurityGroupControllerBase):
 
             # Verify ICMP type and code
             if (ip_protocol.upper() == "ICMP" and
-                (from_port < -1 or to_port > 255)):
+                (from_port < -1 or from_port > 255 or
+                to_port < -1 or to_port > 255)):
                 raise exception.InvalidPortRange(from_port=from_port,
                       to_port=to_port, msg="For ICMP, the"
                                            " type:code must be valid")

nova/tests/api/ec2/test_cloud.py

@@ -382,7 +382,7 @@ class CloudTestCase(test.TestCase):
                                        'userId': u'someuser'}],
                            'ipProtocol': 'udp',
                            'ipRanges': [],
-                           'toPort': 65536},
+                           'toPort': 65535},
                           {'fromPort': 80,
                            'groups': [{'groupName': u'othergroup2',
                                        'userId': u'someuser'}],

nova/tests/api/openstack/compute/contrib/test_security_groups.py

@@ -826,6 +826,42 @@ class TestSecurityGroupRules(test.TestCase):
         self._test_create_with_no_ports_and_no_group('udp')
         self._test_create_with_no_ports('udp')
 
+    def  _test_create_with_ports(self, id_val, proto, from_port, to_port):
+        rule = {
+            'ip_protocol': proto, 'from_port': from_port, 'to_port': to_port,
+            'parent_group_id': '2', 'group_id': '1'
+        }
+        req = fakes.HTTPRequest.blank('/v2/fake/os-security-group-rules')
+        res_dict = self.controller.create(req, {'security_group_rule': rule})
+
+        security_group_rule = res_dict['security_group_rule']
+        expected_rule = {
+            'from_port': from_port,
+            'group': {'tenant_id': '123', 'name': 'test'},
+            'ip_protocol': proto, 'to_port': to_port, 'parent_group_id': 2,
+            'ip_range': {}, 'id': id_val
+        }
+        self.assertTrue(security_group_rule['ip_protocol'] == proto)
+        self.assertTrue(security_group_rule['id'] == id_val)
+        self.assertTrue(security_group_rule['from_port'] == from_port)
+        self.assertTrue(security_group_rule['to_port'] == to_port)
+        self.assertTrue(security_group_rule == expected_rule)
+
+    def test_create_with_ports_icmp(self):
+        self._test_create_with_ports(1, 'icmp', 0, 1)
+        self._test_create_with_ports(2, 'icmp', 0, 0)
+        self._test_create_with_ports(3, 'icmp', 1, 0)
+
+    def test_create_with_ports_tcp(self):
+        self._test_create_with_ports(1, 'tcp', 1, 1)
+        self._test_create_with_ports(2, 'tcp', 1, 65535)
+        self._test_create_with_ports(3, 'tcp', 65535, 65535)
+
+    def test_create_with_ports_udp(self):
+        self._test_create_with_ports(1, 'udp', 1, 1)
+        self._test_create_with_ports(2, 'udp', 1, 65535)
+        self._test_create_with_ports(3, 'udp', 65535, 65535)
+
     def test_delete(self):
         rule = security_group_rule_template(id=10)
 

nova/tests/test_api.py

@@ -394,6 +394,11 @@ class ApiEc2TestCase(test.TestCase):
         group.authorize('tcp', 80, 81, '0.0.0.0/0')
         group.authorize('icmp', -1, -1, '0.0.0.0/0')
         group.authorize('udp', 80, 81, '0.0.0.0/0')
+        group.authorize('tcp', 1, 65535, '0.0.0.0/0')
+        group.authorize('udp', 1, 65535, '0.0.0.0/0')
+        group.authorize('icmp', 1, 0, '0.0.0.0/0')
+        group.authorize('icmp', 0, 1, '0.0.0.0/0')
+        group.authorize('icmp', 0, 0, '0.0.0.0/0')
 
         def _assert(message, *args):
             try:
@@ -414,10 +419,6 @@ class ApiEc2TestCase(test.TestCase):
         _assert('Invalid port range', 'tcp', -1, 1, '0.0.0.0/0')
         # For tcp, valid port range 1-65535
         _assert('Invalid port range', 'tcp', 1, 65599, '0.0.0.0/0')
-        # For icmp, only -1:-1 is allowed for type:code
-        _assert('An unknown error has occurred', 'icmp', -1, 0, '0.0.0.0/0')
-        # Non valid type:code
-        _assert('An unknown error has occurred', 'icmp', 0, 3, '0.0.0.0/0')
         # Invalid Cidr for ICMP type
         _assert('Invalid CIDR', 'icmp', -1, -1, '0.0.444.0/4')
         # Invalid protocol
@@ -441,7 +442,7 @@ class ApiEc2TestCase(test.TestCase):
 
         group = [grp for grp in rv if grp.name == security_group_name][0]
 
-        self.assertEquals(len(group.rules), 3)
+        self.assertEquals(len(group.rules), 8)
         self.assertEquals(int(group.rules[0].from_port), 80)
         self.assertEquals(int(group.rules[0].to_port), 81)
         self.assertEquals(len(group.rules[0].grants), 1)
@@ -454,6 +455,11 @@ class ApiEc2TestCase(test.TestCase):
         group.revoke('tcp', 80, 81, '0.0.0.0/0')
         group.revoke('icmp', -1, -1, '0.0.0.0/0')
         group.revoke('udp', 80, 81, '0.0.0.0/0')
+        group.revoke('tcp', 1, 65535, '0.0.0.0/0')
+        group.revoke('udp', 1, 65535, '0.0.0.0/0')
+        group.revoke('icmp', 1, 0, '0.0.0.0/0')
+        group.revoke('icmp', 0, 1, '0.0.0.0/0')
+        group.revoke('icmp', 0, 0, '0.0.0.0/0')
 
         self.expect_http()
         self.mox.ReplayAll()