<commit_before># Copyright 2012 VMware, Inc.  All rights reserved.
#
#    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.

import netaddr
from neutron_lib.api.definitions import port as port_def
from neutron_lib.api import validators
from neutron_lib.callbacks import events
from neutron_lib.callbacks import exceptions
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib.db import api as db_api
from neutron_lib.db import model_query
from neutron_lib.db import resource_extend
from neutron_lib.db import utils as db_utils
from neutron_lib import exceptions as n_exc
from neutron_lib.utils import helpers
from neutron_lib.utils import net
from oslo_utils import uuidutils
import six
from sqlalchemy.orm import scoped_session

from neutron._i18n import _
from neutron.common import _constants as const
from neutron.common import utils
from neutron.db.models import securitygroup as sg_models
from neutron.db import rbac_db_mixin as rbac_mixin
from neutron.extensions import securitygroup as ext_sg
from neutron.objects import base as base_obj
from neutron.objects import ports as port_obj
from neutron.objects import securitygroup as sg_obj


@resource_extend.has_resource_extenders
@registry.has_registry_receivers
class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase,
                           rbac_mixin.RbacPluginMixin):
    """Mixin class to add security group to db_base_plugin_v2."""

    __native_bulk_support = True

    def create_security_group_bulk(self, context, security_groups):
        return self._create_bulk('security_group', context,
                                 security_groups)

    def _registry_notify(self, res, event, id=None, exc_cls=None, **kwargs):
        # NOTE(armax): a callback exception here will prevent the request
        # from being processed. This is a hook point for backend's validation;
        # we raise to propagate the reason for the failure.
        try:
            if 'payload' in kwargs:
                # TODO(boden): remove shim once all callbacks use payloads
                registry.publish(res, event, self, payload=kwargs['payload'])
            else:
                registry.notify(res, event, self, **kwargs)
        except exceptions.CallbackFailure as e:
            if exc_cls:
                reason = (_('cannot perform %(event)s due to %(reason)s') %
                          {'event': event, 'reason': e})
                raise exc_cls(reason=reason, id=id)

    @db_api.retry_if_session_inactive()
    def create_security_group(self, context, security_group, default_sg=False):
        """Create security group.

        If default_sg is true that means we are a default security group for
        a given tenant if it does not exist.
        """
        s = security_group['security_group']
        kwargs = {
            'context': context,
            'security_group': s,
            'is_default': default_sg,
        }
        self._registry_notify(resources.SECURITY_GROUP, events.BEFORE_CREATE,
                              exc_cls=ext_sg.SecurityGroupConflict,
                              payload=events.DBEventPayload(
                                  context, metadata={'is_default': default_sg},
                                  request_body=security_group,
                                  desired_state=s))

        tenant_id = s['tenant_id']

        if not default_sg:
            self._ensure_default_security_group(context, tenant_id)
        else:
            existing_def_sg_id = self._get_default_sg_id(context, tenant_id)
            if existing_def_sg_id is not None:
                # default already exists, return it
                return self.get_security_group(context, existing_def_sg_id)

        with db_api.CONTEXT_WRITER.using(context):
            sg = sg_obj.SecurityGroup(
                context, id=s.get('id') or uuidutils.generate_uuid(),
                description=s['description'], project_id=tenant_id,
                name=s['name'], is_default=default_sg)
            sg.create()

            for ethertype in ext_sg.sg_supported_ethertypes:
                if default_sg:
                    # Allow intercommunication
                    ingress_rule = sg_obj.SecurityGroupRule(
                        context, id=uuidutils.generate_uuid(),
                        project_id=tenant_id, security_group_id=sg.id,
                        direction='ingress', ethertype=ethertype,
                        remote_group_id=sg.id)
                    ingress_rule.create()
                    sg.rules.append(ingress_rule)

                egress_rule = sg_obj.SecurityGroupRule(
                    context, id=uuidutils.generate_uuid(),
                    project_id=tenant_id, security_group_id=sg.id,
                    direction='egress', ethertype=ethertype)
                egress_rule.create()
                sg.rules.append(egress_rule)
            sg.obj_reset_changes(['rules'])

            # fetch sg from db to load the sg rules with sg model.
            sg = sg_obj.SecurityGroup.get_object(context, id=sg.id)
            secgroup_dict = self._make_security_group_dict(sg)
            kwargs['security_group'] = secgroup_dict
            self._registry_notify(resources.SECURITY_GROUP,
                                  events.PRECOMMIT_CREATE,
                                  exc_cls=ext_sg.SecurityGroupConflict,
                                  **kwargs)

        registry.notify(resources.SECURITY_GROUP, events.AFTER_CREATE, self,
                        **kwargs)
        return secgroup_dict

    @db_api.retry_if_session_inactive()
    def get_security_groups(self, context, filters=None, fields=None,
                            sorts=None, limit=None,
                            marker=None, page_reverse=False, default_sg=False):

        # If default_sg is True do not call _ensure_default_security_group()
        # so this can be done recursively. Context.tenant_id is checked
        # because all the unit tests do not explicitly set the context on
        # GETS. TODO(arosen)  context handling can probably be improved here.
        filters = filters or {}
        if not default_sg and context.tenant_id:
            tenant_id = filters.get('tenant_id')
            if tenant_id:
                tenant_id = tenant_id[0]
            else:
                tenant_id = context.tenant_id
            self._ensure_default_security_group(context, tenant_id)

        pager = base_obj.Pager(
            sorts=sorts, limit=limit, marker=marker, page_reverse=page_reverse)

        sg_objs = sg_obj.SecurityGroup.get_objects(
            context, _pager=pager, validate_filters=False,
            fields=fields, **filters)

        return [self._make_security_group_dict(obj, fields) for obj in sg_objs]

    @db_api.retry_if_session_inactive()
    def get_security_groups_count(self, context, filters=None):
        filters = filters or {}
        return sg_obj.SecurityGroup.count(
            context, validate_filters=False, **filters)

    @db_api.retry_if_session_inactive()
    def get_security_group(self, context, id, fields=None, tenant_id=None):
        """Tenant id is given to handle the case when creating a security
        group rule on behalf of another use.
        """
        if tenant_id:
            tmp_context_tenant_id = context.tenant_id
            context.tenant_id = tenant_id

        try:
            with db_api.CONTEXT_READER.using(context):
                ret = self._make_security_group_dict(self._get_security_group(
                                                     context, id,
                                                     fields=fields),
                                                     fields)
                if (fields is None or len(fields) == 0 or
                   'security_group_rules' in fields):
                    rules = self.get_security_group_rules(
                        context, {'security_group_id': [id]})
                    ret['security_group_rules'] = rules

        finally:
            if tenant_id:
                context.tenant_id = tmp_context_tenant_id
        return ret

    def _get_security_group(self, context, id, fields=None):
        sg = sg_obj.SecurityGroup.get_object(context, fields=fields, id=id)
        if sg is None:
            raise ext_sg.SecurityGroupNotFound(id=id)
        return sg

    def _check_security_group(self, context, id, tenant_id=None):
        if tenant_id:
            tmp_context_tenant_id = context.tenant_id
            context.tenant_id = tenant_id

        try:
            if not sg_obj.SecurityGroup.objects_exist(context, id=id):
                raise ext_sg.SecurityGroupNotFound(id=id)
        finally:
            if tenant_id:
                context.tenant_id = tmp_context_tenant_id

    @db_api.retry_if_session_inactive()
    def delete_security_group(self, context, id):<commit_msg>Show all SG rules belong to SG in group's details<commit_after># Copyright 2012 VMware, Inc.  All rights reserved.
#
#    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.

import netaddr
from neutron_lib.api.definitions import port as port_def
from neutron_lib.api import validators
from neutron_lib.callbacks import events
from neutron_lib.callbacks import exceptions
from neutron_lib.callbacks import registry
from neutron_lib.callbacks import resources
from neutron_lib import constants
from neutron_lib import context as context_lib
from neutron_lib.db import api as db_api
from neutron_lib.db import model_query
from neutron_lib.db import resource_extend
from neutron_lib.db import utils as db_utils
from neutron_lib import exceptions as n_exc
from neutron_lib.utils import helpers
from neutron_lib.utils import net
from oslo_utils import uuidutils
import six
from sqlalchemy.orm import scoped_session

from neutron._i18n import _
from neutron.common import _constants as const
from neutron.common import utils
from neutron.db.models import securitygroup as sg_models
from neutron.db import rbac_db_mixin as rbac_mixin
from neutron.extensions import securitygroup as ext_sg
from neutron.objects import base as base_obj
from neutron.objects import ports as port_obj
from neutron.objects import securitygroup as sg_obj


@resource_extend.has_resource_extenders
@registry.has_registry_receivers
class SecurityGroupDbMixin(ext_sg.SecurityGroupPluginBase,
                           rbac_mixin.RbacPluginMixin):
    """Mixin class to add security group to db_base_plugin_v2."""

    __native_bulk_support = True

    def create_security_group_bulk(self, context, security_groups):
        return self._create_bulk('security_group', context,
                                 security_groups)

    def _registry_notify(self, res, event, id=None, exc_cls=None, **kwargs):
        # NOTE(armax): a callback exception here will prevent the request
        # from being processed. This is a hook point for backend's validation;
        # we raise to propagate the reason for the failure.
        try:
            if 'payload' in kwargs:
                # TODO(boden): remove shim once all callbacks use payloads
                registry.publish(res, event, self, payload=kwargs['payload'])
            else:
                registry.notify(res, event, self, **kwargs)
        except exceptions.CallbackFailure as e:
            if exc_cls:
                reason = (_('cannot perform %(event)s due to %(reason)s') %
                          {'event': event, 'reason': e})
                raise exc_cls(reason=reason, id=id)

    @db_api.retry_if_session_inactive()
    def create_security_group(self, context, security_group, default_sg=False):
        """Create security group.

        If default_sg is true that means we are a default security group for
        a given tenant if it does not exist.
        """
        s = security_group['security_group']
        kwargs = {
            'context': context,
            'security_group': s,
            'is_default': default_sg,
        }
        self._registry_notify(resources.SECURITY_GROUP, events.BEFORE_CREATE,
                              exc_cls=ext_sg.SecurityGroupConflict,
                              payload=events.DBEventPayload(
                                  context, metadata={'is_default': default_sg},
                                  request_body=security_group,
                                  desired_state=s))

        tenant_id = s['tenant_id']

        if not default_sg:
            self._ensure_default_security_group(context, tenant_id)
        else:
            existing_def_sg_id = self._get_default_sg_id(context, tenant_id)
            if existing_def_sg_id is not None:
                # default already exists, return it
                return self.get_security_group(context, existing_def_sg_id)

        with db_api.CONTEXT_WRITER.using(context):
            sg = sg_obj.SecurityGroup(
                context, id=s.get('id') or uuidutils.generate_uuid(),
                description=s['description'], project_id=tenant_id,
                name=s['name'], is_default=default_sg)
            sg.create()

            for ethertype in ext_sg.sg_supported_ethertypes:
                if default_sg:
                    # Allow intercommunication
                    ingress_rule = sg_obj.SecurityGroupRule(
                        context, id=uuidutils.generate_uuid(),
                        project_id=tenant_id, security_group_id=sg.id,
                        direction='ingress', ethertype=ethertype,
                        remote_group_id=sg.id)
                    ingress_rule.create()
                    sg.rules.append(ingress_rule)

                egress_rule = sg_obj.SecurityGroupRule(
                    context, id=uuidutils.generate_uuid(),
                    project_id=tenant_id, security_group_id=sg.id,
                    direction='egress', ethertype=ethertype)
                egress_rule.create()
                sg.rules.append(egress_rule)
            sg.obj_reset_changes(['rules'])

            # fetch sg from db to load the sg rules with sg model.
            sg = sg_obj.SecurityGroup.get_object(context, id=sg.id)
            secgroup_dict = self._make_security_group_dict(sg)
            kwargs['security_group'] = secgroup_dict
            self._registry_notify(resources.SECURITY_GROUP,
                                  events.PRECOMMIT_CREATE,
                                  exc_cls=ext_sg.SecurityGroupConflict,
                                  **kwargs)

        registry.notify(resources.SECURITY_GROUP, events.AFTER_CREATE, self,
                        **kwargs)
        return secgroup_dict

    @db_api.retry_if_session_inactive()
    def get_security_groups(self, context, filters=None, fields=None,
                            sorts=None, limit=None,
                            marker=None, page_reverse=False, default_sg=False):

        # If default_sg is True do not call _ensure_default_security_group()
        # so this can be done recursively. Context.tenant_id is checked
        # because all the unit tests do not explicitly set the context on
        # GETS. TODO(arosen)  context handling can probably be improved here.
        filters = filters or {}
        if not default_sg and context.tenant_id:
            tenant_id = filters.get('tenant_id')
            if tenant_id:
                tenant_id = tenant_id[0]
            else:
                tenant_id = context.tenant_id
            self._ensure_default_security_group(context, tenant_id)

        pager = base_obj.Pager(
            sorts=sorts, limit=limit, marker=marker, page_reverse=page_reverse)

        sg_objs = sg_obj.SecurityGroup.get_objects(
            context, _pager=pager, validate_filters=False,
            fields=fields, **filters)

        return [self._make_security_group_dict(obj, fields) for obj in sg_objs]

    @db_api.retry_if_session_inactive()
    def get_security_groups_count(self, context, filters=None):
        filters = filters or {}
        return sg_obj.SecurityGroup.count(
            context, validate_filters=False, **filters)

    @db_api.retry_if_session_inactive()
    def get_security_group(self, context, id, fields=None, tenant_id=None):
        """Tenant id is given to handle the case when creating a security
        group rule on behalf of another use.
        """
        if tenant_id:
            tmp_context_tenant_id = context.tenant_id
            context.tenant_id = tenant_id

        try:
            with db_api.CONTEXT_READER.using(context):
                ret = self._make_security_group_dict(self._get_security_group(
                                                     context, id,
                                                     fields=fields),
                                                     fields)
                if (fields is None or len(fields) == 0 or
                   'security_group_rules' in fields):
                    rules = self.get_security_group_rules(
                        context_lib.get_admin_context(),
                        {'security_group_id': [id]})
                    ret['security_group_rules'] = rules

        finally:
            if tenant_id:
                context.tenant_id = tmp_context_tenant_id
        return ret

    def _get_security_group(self, context, id, fields=None):
        sg = sg_obj.SecurityGroup.get_object(context, fields=fields, id=id)
        if sg is None:
            raise ext_sg.SecurityGroupNotFound(id=id)
        return sg

    def _check_security_group(self, context, id, tenant_id=None):
        if tenant_id:
            tmp_context_tenant_id = context.tenant_id
            context.tenant_id = tenant_id

        try:
            if not sg_obj.SecurityGroup.objects_exist(context, id=id):
                raise ext_sg.SecurityGroupNotFound(id=id)
        finally:
            if tenant_id:
                context.tenant_id = tmp_context_tenant_id

    @db_api.retry_if_session_inactive()
    def delete_security_group(self, context, id):
{'fraction_non_alphanumeric': 0.0683161583930533, 'fraction_numerical': 0.001360046032327248, 'mean_word_length': 2.1647078298294984, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 26, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '3374295', 'n_tokens_mistral': 5189, 'n_tokens_neox': 5007, 'n_words': 1352}