CVE-2012-2101 (Python, published 2012-06-07T19:55Z)
Description: value: Openstack Compute (Nova) Folsom, 2012.1, and 2011.3 does not limit the number of security group rules, which allows remote authenticated users with certain permissions to cause a denial of service (CPU and hard drive consumption) via a network request that triggers a large number of iptables rules.
Severity: LOW (CVSS v3 base score: )
Weakness: CWE-264 \u2014 Permissions, Privileges, and Access Controls
Weaknesses in this category are related to the management of permissions, privileges, and other security features that are used to perform access control.
Repository: https://github.com/openstack/nova
Files changed:
nova/api/ec2/cloud.py
nova/api/openstack/compute/contrib/security_groups.py
nova/db/api.py
nova/db/sqlalchemy/api.py
nova/quota.py
nova/tests/api/ec2/test_cloud.py
nova/tests/api/openstack/compute/contrib/test_quota_classes.py
nova/tests/api/openstack/compute/contrib/test_quotas.py
nova/tests/api/openstack/compute/contrib/test_security_groups.py
nova/tests/test_quota.py
Commit message: Implement security group quotas.

Fixes LP Bug #969545

Change-Id: I60d4300aa04597e2d8b0eea31ab0303b0b3c48f9


Vulnerable code:
        'injected_files', 'cores']
            'cores': 20, 'injected_file_content_bytes': 10240}}
            'injected_file_content_bytes': 10240}
                                    'injected_file_content_bytes': 10240}}
                                    'injected_file_content_bytes': 10240}}
            'injected_file_content_bytes': 10240}}
            'injected_file_content_bytes': 10240}
                    'injected_file_content_bytes': 10240}}
                              'injected_file_content_bytes': 10240}}
                              'injected_file_content_bytes': 10240}}

Fixed code:
from nova import quota
        allowed = quota.allowed_security_group_rules(context,
                                                   security_group['id'],
                                                   1)
        if allowed < 1:
            msg = _("Quota exceeded, too many security group rules.")
            raise exception.EC2APIError(msg)

        if quota.allowed_security_groups(context, 1) < 1:
            msg = _("Quota exceeded, too many security groups.")
            raise exception.EC2APIError(msg)

from nova import quota
        if quota.allowed_security_groups(context, 1) < 1:
            msg = _("Quota exceeded, too many security groups.")
            raise exc.HTTPBadRequest(explanation=msg)

        allowed = quota.allowed_security_group_rules(context,
                                                   parent_group_id,
                                                   1)
        if allowed < 1:
            msg = _("Quota exceeded, too many security group rules.")
            raise exc.HTTPBadRequest(explanation=msg)

def security_group_count_by_project(context, project_id):
    """Count number of security groups in a project."""
    return IMPL.security_group_count_by_project(context, project_id)


def security_group_rule_count_by_group(context, security_group_id):
    """Count rules in a given security group."""
    return IMPL.security_group_rule_count_by_group(context, security_group_id)


@require_context
def security_group_count_by_project(context, project_id):
    authorize_project_context(context, project_id)
    return model_query(context, models.SecurityGroup, read_deleted="no").\
                   filter_by(project_id=project_id).\
                   count()

@require_context
def security_group_rule_count_by_group(context, security_group_id):
    return model_query(context, models.SecurityGroupIngressRule,
                   read_deleted="no").\
                   filter_by(parent_group_id=security_group_id).\
                   count()

#
    cfg.IntOpt('quota_security_groups',
               default=10,
               help='number of security groups per project'),
    cfg.IntOpt('quota_security_group_rules',
               default=20,
               help='number of security rules per security group'),
        'injected_files', 'cores', 'security_groups', 'security_group_rules']
        'security_groups': FLAGS.quota_security_groups,
        'security_group_rules': FLAGS.quota_security_group_rules,
def allowed_security_groups(context, requested_security_groups):
    """Check quota and return min(requested, allowed) security groups."""
    project_id = context.project_id
    context = context.elevated()
    used_sec_groups = db.security_group_count_by_project(context, project_id)
    quota = get_project_quotas(context, project_id)
    allowed_sec_groups = _get_request_allotment(requested_security_groups,
                                                  used_sec_groups,
                                                  quota['security_groups'])
    return min(requested_security_groups, allowed_sec_groups)


def allowed_security_group_rules(context, security_group_id,
        requested_rules):
    """Check quota and return min(requested, allowed) sec group rules."""
    project_id = context.project_id
    context = context.elevated()
    used_rules = db.security_group_rule_count_by_group(context,
                                                            security_group_id)
    quota = get_project_quotas(context, project_id)
    allowed_rules = _get_request_allotment(requested_rules,
                                              used_rules,
                                              quota['security_group_rules'])
    return min(requested_rules, allowed_rules)


    def test_security_group_quota_limit(self):
        self.flags(quota_security_groups=10)
        for i in range(1, 10):
            name = 'test name %i' % i
            descript = 'test description %i' % i
            create = self.cloud.create_security_group
            result = create(self.context, name, descript)

        # 11'th group should fail
        self.assertRaises(exception.EC2APIError,
                          create, self.context, 'foo', 'bar')

    def test_security_group_ingress_quota_limit(self):
        self.flags(quota_security_group_rules=20)
        kwargs = {'project_id': self.context.project_id, 'name': 'test'}
        sec_group = db.security_group_create(self.context, kwargs)
        authz = self.cloud.authorize_security_group_ingress
        for i in range(100, 120):
            kwargs = {'to_port': i, 'from_port': i, 'ip_protocol': 'tcp'}
            authz(self.context, group_id=sec_group['id'], **kwargs)

        kwargs = {'to_port': 121, 'from_port': 121, 'ip_protocol': 'tcp'}
        self.assertRaises(exception.EC2APIError, authz, self.context,
                              group_id=sec_group['id'], **kwargs)

            'cores': 20, 'injected_file_content_bytes': 10240,
            'security_groups': 10, 'security_group_rules': 20}}
            'injected_file_content_bytes': 10240,
            'security_groups': 10,
            'security_group_rules': 20,
            }
        self.assertEqual(qs['security_groups'], 10)
        self.assertEqual(qs['security_group_rules'], 20)
                                    'injected_file_content_bytes': 10240,
                                    'security_groups': 10,
                                    'security_group_rules': 20,
                                    }}
                                    'injected_file_content_bytes': 10240,
                                    'security_groups': 10,
                                    'security_group_rules': 20,
                                    }}
                security_groups=10,
                security_group_rules=20,
                security_groups='10',
                security_group_rules='20',
                  '<security_groups>10</security_groups>'
                  '<security_group_rules>20</security_group_rules>'
            'injected_file_content_bytes': 10240,
            'security_groups': 10, 'security_group_rules': 20}}
            'injected_file_content_bytes': 10240,
            'security_groups': 10,
            'security_group_rules': 20,
            }
        self.assertEqual(qs['security_groups'], 10)
        self.assertEqual(qs['security_group_rules'], 20)
                    'injected_file_content_bytes': 10240,
                    'security_groups': 10,
                    'security_group_rules': 20,
                    }}
                              'injected_file_content_bytes': 10240,
                              'security_groups': 10,
                              'security_group_rules': 20}}
                              'injected_file_content_bytes': 10240,
                              'security_groups': 10,
                              'security_group_rules': 20}}
                security_groups=10,
                security_group_rules=20,
                security_groups='10',
                security_group_rules='20',
                  '<security_groups>10</security_groups>'
                  '<security_group_rules>20</security_group_rules>'
from nova import flags
FLAGS = flags.FLAGS

    def test_create_security_group_quota_limit(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-groups')
        for num in range(1, FLAGS.quota_security_groups):
            name = 'test%s' % num
            sg = security_group_template(name=name)
            res_dict = self.controller.create(req, {'security_group': sg})
            self.assertEqual(res_dict['security_group']['name'], name)

        sg = security_group_template()
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group': sg})

    def test_create_rule_quota_limit(self):
        req = fakes.HTTPRequest.blank('/v2/fake/os-security-group-rules')
        for num in range(100, 100 + FLAGS.quota_security_group_rules):
            rule = {
                'ip_protocol': 'tcp', 'from_port': num,
                'to_port': num, 'parent_group_id': '2', 'group_id': '1'
            }
            self.controller.create(req, {'security_group_rule': rule})

        rule = {
            'ip_protocol': 'tcp', 'from_port': '121', 'to_port': '121',
            'parent_group_id': '2', 'group_id': '1'
        }
        self.assertRaises(webob.exc.HTTPBadRequest, self.controller.create,
                          req, {'security_group_rule': rule})

                   quota_security_groups=10,
                   quota_security_group_rules=20,
                    quota_security_groups=10,
                    quota_security_group_rules=20,
                    security_groups=5,
                    security_group_rules=10,
                security_groups=10,
                security_group_rules=20,
                   quota_security_groups=-1,
                   quota_security_group_rules=-1,
                security_groups=-1,
                security_group_rules=-1,
                security_groups=10,
                security_group_rules=20,
                security_groups=10,
                security_group_rules=20,
                security_groups=10,
                security_group_rules=20,
                security_groups=5,
                security_group_rules=10,
                security_groups=10,
                security_group_rules=20,
                security_groups=5,
                security_group_rules=10,
    def test_unlimited_security_groups(self):
        self.flags(quota_security_groups=10)
        security_groups = quota.allowed_security_groups(self.context, 100)
        self.assertEqual(security_groups, 10)
        db.quota_create(self.context, self.project_id, 'security_groups', -1)
        security_groups = quota.allowed_security_groups(self.context, 100)
        self.assertEqual(security_groups, 100)
        security_groups = quota.allowed_security_groups(self.context, 101)
        self.assertEqual(security_groups, 101)

    def test_unlimited_security_group_rules(self):

        def fake_security_group_rule_count_by_group(context, sec_group_id):
            return 0

        self.stubs.Set(db, 'security_group_rule_count_by_group',
                       fake_security_group_rule_count_by_group)

        self.flags(quota_security_group_rules=20)
        rules = quota.allowed_security_group_rules(self.context, 1234, 100)
        self.assertEqual(rules, 20)
        db.quota_create(self.context, self.project_id, 'security_group_rules',
                        -1)
        rules = quota.allowed_security_group_rules(self.context, 1234, 100)
        self.assertEqual(rules, 100)
        rules = quota.allowed_security_group_rules(self.context, 1234, 101)
        self.assertEqual(rules, 101)