Fix issues with security group auths without ports.

Fix bug 946427.

There was a bug where a security group would get completely opened in
cases where only icmp, udp, or tcp should be opened.  For example, any
of the following three commands would result in opening everything:

    euca-authorize -P icmp -o test-ports test-ports
    euca-authorize -P tcp -o test-ports test-ports
    euca-authorize -P udp -o test-ports test-ports

This patch resolves this and these commands now only open the protocol
that was specified.  Unit tests have been added to verify the fix and
also verify that this only works when a source group is specified.
While the bug was originally reported against the EC2 API, the same
updates and similar unit tests have gone in to the equivalent code for
the OpenStack API.

Change-Id: I4c87c5f5f4ccee60c6c16da4e659d73ab3f4a34f

2 files changed, 90 insertions, 0 deletions

diff --git a/nova/tests/api/ec2/test_cloud.py b/nova/tests/api/ec2/test_cloud.py
index 5a64f237e..b9d9c7136 100644
--- a/nova/tests/api/ec2/test_cloud.py
+++ b/nova/tests/api/ec2/test_cloud.py
@@ -434,6 +434,60 @@ class CloudTestCase(test.TestCase):
         self.assertRaises(exception.EC2APIError, authz, self.context,
                           group_name=sec['name'], **kwargs)
 
+    def _test_authorize_security_group_no_ports_with_source_group(self, proto):
+        kwargs = {'project_id': self.context.project_id, 'name': 'test'}
+        sec = db.security_group_create(self.context, kwargs)
+
+        authz = self.cloud.authorize_security_group_ingress
+        auth_kwargs = {'ip_protocol': proto,
+                       'groups': {'1': {'user_id': self.context.user_id,
+                                        'group_name': u'test'}}}
+        self.assertTrue(authz(self.context, group_name=sec['name'],
+                        **auth_kwargs))
+
+        describe = self.cloud.describe_security_groups
+        groups = describe(self.context, group_name=['test'])
+        self.assertEquals(len(groups['securityGroupInfo']), 1)
+        actual_rules = groups['securityGroupInfo'][0]['ipPermissions']
+        expected_rules = [{'groups': [{'groupName': 'test',
+                                       'userId': self.context.user_id}],
+                           'ipProtocol': proto,
+                           'ipRanges': []}]
+        if proto == 'icmp':
+            expected_rules[0]['fromPort'] = -1
+            expected_rules[0]['toPort'] = -1
+        else:
+            expected_rules[0]['fromPort'] = 1
+            expected_rules[0]['toPort'] = 65535
+        self.assertTrue(expected_rules == actual_rules)
+        describe = self.cloud.describe_security_groups
+        groups = describe(self.context, group_name=['test'])
+
+        db.security_group_destroy(self.context, sec['id'])
+
+    def _test_authorize_security_group_no_ports_no_source_group(self, proto):
+        kwargs = {'project_id': self.context.project_id, 'name': 'test'}
+        sec = db.security_group_create(self.context, kwargs)
+
+        authz = self.cloud.authorize_security_group_ingress
+        auth_kwargs = {'ip_protocol': proto}
+        self.assertRaises(exception.EC2APIError, authz, self.context,
+                          group_name=sec['name'], **auth_kwargs)
+
+        db.security_group_destroy(self.context, sec['id'])
+
+    def test_authorize_security_group_no_ports_icmp(self):
+        self._test_authorize_security_group_no_ports_with_source_group('icmp')
+        self._test_authorize_security_group_no_ports_no_source_group('icmp')
+
+    def test_authorize_security_group_no_ports_tcp(self):
+        self._test_authorize_security_group_no_ports_with_source_group('tcp')
+        self._test_authorize_security_group_no_ports_no_source_group('tcp')
+
+    def test_authorize_security_group_no_ports_udp(self):
+        self._test_authorize_security_group_no_ports_with_source_group('udp')
+        self._test_authorize_security_group_no_ports_no_source_group('udp')
+
     def test_revoke_security_group_ingress_missing_group_name_or_id(self):
         kwargs = {'to_port': '999', 'from_port': '999', 'ip_protocol': 'tcp'}
         revoke = self.cloud.revoke_security_group_ingress
diff --git a/nova/tests/api/openstack/compute/contrib/test_security_groups.py b/nova/tests/api/openstack/compute/contrib/test_security_groups.py
index 782eb409b..f1d86c0e6 100644
--- a/nova/tests/api/openstack/compute/contrib/test_security_groups.py
+++ b/nova/tests/api/openstack/compute/contrib/test_security_groups.py
@@ -790,6 +790,42 @@ class TestSecurityGroupRules(test.TestCase):
         self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                           req, {'security_group_rule': rule})
 
+    def _test_create_with_no_ports_and_no_group(self, proto):
+        rule = {'ip_protocol': proto, 'parent_group_id': '2'}
+
+        req = fakes.HTTPRequest.blank('/v2/fake/os-security-group-rules')
+        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
+                          req, {'security_group_rule': rule})
+
+    def  _test_create_with_no_ports(self, proto):
+        rule = {'ip_protocol': proto, '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': 1, 'group': {'tenant_id': '123', 'name': 'test'},
+            'ip_protocol': proto, 'to_port': 65535, 'parent_group_id': 2,
+            'ip_range': {}, 'id': 1
+        }
+        if proto == 'icmp':
+            expected_rule['to_port'] = -1
+            expected_rule['from_port'] = -1
+        self.assertTrue(security_group_rule == expected_rule)
+
+    def test_create_with_no_ports_icmp(self):
+        self._test_create_with_no_ports_and_no_group('icmp')
+        self._test_create_with_no_ports('icmp')
+
+    def test_create_with_no_ports_tcp(self):
+        self._test_create_with_no_ports_and_no_group('tcp')
+        self._test_create_with_no_ports('tcp')
+
+    def test_create_with_no_ports_udp(self):
+        self._test_create_with_no_ports_and_no_group('udp')
+        self._test_create_with_no_ports('udp')
+
     def test_delete(self):
         rule = security_group_rule_template(id=10)