<commit_before># Copyright (c) 2012 OpenStack Foundation.
# 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.

from collections import abc
import itertools
import re
import sys

from neutron_lib.api import attributes
from neutron_lib.api.definitions import network as net_apidef
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from neutron_lib.services import constants as service_const
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log as logging
from oslo_policy import opts
from oslo_policy import policy
from oslo_utils import excutils
import stevedore

from neutron._i18n import _
from neutron.common import cache_utils as cache


LOG = logging.getLogger(__name__)

_ENFORCER = None
ADMIN_CTX_POLICY = 'context_is_admin'
ADVSVC_CTX_POLICY = 'context_is_advsvc'

# Identify the attribute used by a resource to reference another resource
_RESOURCE_FOREIGN_KEYS = {
    net_apidef.COLLECTION_NAME: 'network_id',
    # TODO(slaweq): use SECURITYGROUPS constant from api def when
    # securitygroups api def will be moved to neutron-lib
    'security_groups': 'security_group_id'
}


# TODO(gmann): Remove setting the default value of config policy_file
# once oslo_policy change the default value to 'policy.yaml'.
# https://github.com/openstack/oslo.policy/blob/a626ad12fe5a3abd49d70e3e5b95589d279ab578/oslo_policy/opts.py#L49
DEFAULT_POLICY_FILE = 'policy.yaml'
opts.set_defaults(cfg.CONF, DEFAULT_POLICY_FILE)


def reset():
    global _ENFORCER
    if _ENFORCER:
        _ENFORCER.clear()
        _ENFORCER = None


def register_rules(enforcer):
    extmgr = stevedore.extension.ExtensionManager('neutron.policies',
                                                  invoke_on_load=True)
    policies = [list(e.obj) for e in extmgr.extensions]
    LOG.debug('Loaded default policies from %s '
              'under neutron.policies entry points',
              [e.name for e in extmgr.extensions])
    enforcer.register_defaults(itertools.chain(*policies))


def init(conf=cfg.CONF, policy_file=None, suppress_deprecation_warnings=False):
    """Init an instance of the Enforcer class."""

    global _ENFORCER
    if not _ENFORCER:
        _ENFORCER = policy.Enforcer(conf, policy_file=policy_file)
        # TODO(slaweq) Explictly disable the warnings for policies
        # changing their default check_str. During policy-defaults-refresh
        # work, all the policy defaults have been changed and warning for
        # each policy started filling the logs limit for various tool.
        # Once we move to new defaults only world then we can enable these
        # warning again.
        _ENFORCER.suppress_default_change_warnings = True
        if suppress_deprecation_warnings:
            _ENFORCER.suppress_deprecation_warnings = True
        register_rules(_ENFORCER)
        _ENFORCER.load_rules(True)


def refresh(policy_file=None):
    """Reset policy and init a new instance of Enforcer."""
    reset()
    init(policy_file=policy_file)


def get_resource_and_action(action, pluralized=None):
    """Return resource and enforce_attr_based_check(boolean) per
       resource and action extracted from api operation.
    """
    data = action.split(':', 1)[0].split('_', 1)
    resource = pluralized or ("%ss" % data[-1])
    enforce_attr_based_check = data[0] not in ('get', 'delete')
    return (resource, enforce_attr_based_check)


def set_rules(policies, overwrite=True):
    """Set rules based on the provided dict of rules.

    :param policies: New policies to use. It should be an instance of dict.
    :param overwrite: Whether to overwrite current rules or update them
                          with the new rules.
    """

    LOG.debug("Loading policies from file: %s", _ENFORCER.policy_path)
    init()
    _ENFORCER.set_rules(policies, overwrite)


def _is_attribute_explicitly_set(attribute_name, resource, target, action):
    """Verify that an attribute is present and is explicitly set."""
    if target.get(constants.ATTRIBUTES_TO_UPDATE):
        # In the case of update, the function should not pay attention to a
        # default value of an attribute, but check whether it was explicitly
        # marked as being updated instead.
        return (attribute_name in target[constants.ATTRIBUTES_TO_UPDATE] and
                target[attribute_name] is not constants.ATTR_NOT_SPECIFIED)
    result = (attribute_name in target and
              target[attribute_name] is not constants.ATTR_NOT_SPECIFIED)
    if result and 'default' in resource[attribute_name]:
        return target[attribute_name] != resource[attribute_name]['default']
    return result


def _should_validate_sub_attributes(attribute, sub_attr):
    """Verify that sub-attributes are iterable and should be validated."""
    validate = attribute.get('validate')
    return (validate and isinstance(sub_attr, abc.Iterable) and
            any([k.startswith('type:dict') and
                 v for (k, v) in validate.items()]))


def _build_subattr_match_rule(attr_name, attr, action, target):
    """Create the rule to match for sub-attribute policy checks."""
    # TODO(salv-orlando): Instead of relying on validator info, introduce
    # typing for API attributes
    # Expect a dict as type descriptor
    validate = attr['validate']
    key = [k for k in validate.keys() if k.startswith('type:dict')]
    if not key:
        LOG.warning("Unable to find data type descriptor for attribute %s",
                    attr_name)
        return
    data = validate[key[0]]
    if not isinstance(data, dict):
        LOG.debug("Attribute type descriptor is not a dict. Unable to "
                  "generate any sub-attr policy rule for %s.",
                  attr_name)
        return
    sub_attr_rules = [policy.RuleCheck('rule', '%s:%s:%s' %
                                       (action, attr_name,
                                        sub_attr_name)) for
                      sub_attr_name in data if sub_attr_name in
                      target[attr_name]]
    return policy.AndCheck(sub_attr_rules)


def _build_list_of_subattrs_rule(attr_name, attribute_value, action):
    rules = []
    for sub_attr in attribute_value:
        if isinstance(sub_attr, dict):
            for k in sub_attr:
                rules.append(policy.RuleCheck(
                    'rule', '%s:%s:%s' % (action, attr_name, k)))
    if rules:
        return policy.AndCheck(rules)


def _process_rules_list(rules, match_rule):
    """Recursively walk a policy rule to extract a list of match entries."""
    if isinstance(match_rule, policy.RuleCheck):
        rules.append(match_rule.match)
    elif isinstance(match_rule, policy.AndCheck):
        for rule in match_rule.rules:
            _process_rules_list(rules, rule)
    return rules


def _build_match_rule(action, target, pluralized):
    """Create the rule to match for a given action.

    The policy rule to be matched is built in the following way:
    1) add entries for matching permission on objects
    2) add an entry for the specific action (e.g.: create_network)
    3) add an entry for attributes of a resource for which the action
       is being executed (e.g.: create_network:shared)
    4) add an entry for sub-attributes of a resource for which the
       action is being executed
       (e.g.: create_router:external_gateway_info:network_id)
    """
    match_rule = policy.RuleCheck('rule', action)
    resource, enforce_attr_based_check = get_resource_and_action(
        action, pluralized)
    if enforce_attr_based_check:
        # assigning to variable with short name for improving readability<commit_msg>[Policy] Set scope_types for BaseCheck rules<commit_after># Copyright (c) 2012 OpenStack Foundation.
# 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.

from collections import abc
import itertools
import re
import sys

from neutron_lib.api import attributes
from neutron_lib.api.definitions import network as net_apidef
from neutron_lib import constants
from neutron_lib import context
from neutron_lib import exceptions
from neutron_lib.plugins import directory
from neutron_lib.services import constants as service_const
from oslo_config import cfg
from oslo_db import exception as db_exc
from oslo_log import log as logging
from oslo_policy import opts
from oslo_policy import policy
from oslo_utils import excutils
import stevedore

from neutron._i18n import _
from neutron.common import cache_utils as cache


LOG = logging.getLogger(__name__)

_ENFORCER = None
ADMIN_CTX_POLICY = 'context_is_admin'
ADVSVC_CTX_POLICY = 'context_is_advsvc'

# Identify the attribute used by a resource to reference another resource
_RESOURCE_FOREIGN_KEYS = {
    net_apidef.COLLECTION_NAME: 'network_id',
    # TODO(slaweq): use SECURITYGROUPS constant from api def when
    # securitygroups api def will be moved to neutron-lib
    'security_groups': 'security_group_id'
}


# TODO(gmann): Remove setting the default value of config policy_file
# once oslo_policy change the default value to 'policy.yaml'.
# https://github.com/openstack/oslo.policy/blob/a626ad12fe5a3abd49d70e3e5b95589d279ab578/oslo_policy/opts.py#L49
DEFAULT_POLICY_FILE = 'policy.yaml'
opts.set_defaults(cfg.CONF, DEFAULT_POLICY_FILE)


def reset():
    global _ENFORCER
    if _ENFORCER:
        _ENFORCER.clear()
        _ENFORCER = None


def register_rules(enforcer):
    extmgr = stevedore.extension.ExtensionManager('neutron.policies',
                                                  invoke_on_load=True)
    policies = [list(e.obj) for e in extmgr.extensions]
    LOG.debug('Loaded default policies from %s '
              'under neutron.policies entry points',
              [e.name for e in extmgr.extensions])
    enforcer.register_defaults(itertools.chain(*policies))


def init(conf=cfg.CONF, policy_file=None, suppress_deprecation_warnings=False):
    """Init an instance of the Enforcer class."""

    global _ENFORCER
    if not _ENFORCER:
        _ENFORCER = policy.Enforcer(conf, policy_file=policy_file)
        # TODO(slaweq) Explictly disable the warnings for policies
        # changing their default check_str. During policy-defaults-refresh
        # work, all the policy defaults have been changed and warning for
        # each policy started filling the logs limit for various tool.
        # Once we move to new defaults only world then we can enable these
        # warning again.
        _ENFORCER.suppress_default_change_warnings = True
        if suppress_deprecation_warnings:
            _ENFORCER.suppress_deprecation_warnings = True
        register_rules(_ENFORCER)
        _ENFORCER.load_rules(True)


def refresh(policy_file=None):
    """Reset policy and init a new instance of Enforcer."""
    reset()
    init(policy_file=policy_file)


def get_resource_and_action(action, pluralized=None):
    """Return resource and enforce_attr_based_check(boolean) per
       resource and action extracted from api operation.
    """
    data = action.split(':', 1)[0].split('_', 1)
    resource = pluralized or ("%ss" % data[-1])
    enforce_attr_based_check = data[0] not in ('get', 'delete')
    return (resource, enforce_attr_based_check)


def set_rules(policies, overwrite=True):
    """Set rules based on the provided dict of rules.

    :param policies: New policies to use. It should be an instance of dict.
    :param overwrite: Whether to overwrite current rules or update them
                          with the new rules.
    """

    LOG.debug("Loading policies from file: %s", _ENFORCER.policy_path)
    init()
    _ENFORCER.set_rules(policies, overwrite)


def _is_attribute_explicitly_set(attribute_name, resource, target, action):
    """Verify that an attribute is present and is explicitly set."""
    if target.get(constants.ATTRIBUTES_TO_UPDATE):
        # In the case of update, the function should not pay attention to a
        # default value of an attribute, but check whether it was explicitly
        # marked as being updated instead.
        return (attribute_name in target[constants.ATTRIBUTES_TO_UPDATE] and
                target[attribute_name] is not constants.ATTR_NOT_SPECIFIED)
    result = (attribute_name in target and
              target[attribute_name] is not constants.ATTR_NOT_SPECIFIED)
    if result and 'default' in resource[attribute_name]:
        return target[attribute_name] != resource[attribute_name]['default']
    return result


def _should_validate_sub_attributes(attribute, sub_attr):
    """Verify that sub-attributes are iterable and should be validated."""
    validate = attribute.get('validate')
    return (validate and isinstance(sub_attr, abc.Iterable) and
            any([k.startswith('type:dict') and
                 v for (k, v) in validate.items()]))


def _build_subattr_match_rule(attr_name, attr, action, target):
    """Create the rule to match for sub-attribute policy checks."""
    # TODO(salv-orlando): Instead of relying on validator info, introduce
    # typing for API attributes
    # Expect a dict as type descriptor
    validate = attr['validate']
    key = [k for k in validate.keys() if k.startswith('type:dict')]
    if not key:
        LOG.warning("Unable to find data type descriptor for attribute %s",
                    attr_name)
        return
    data = validate[key[0]]
    if not isinstance(data, dict):
        LOG.debug("Attribute type descriptor is not a dict. Unable to "
                  "generate any sub-attr policy rule for %s.",
                  attr_name)
        return
    sub_attr_rules = [policy.RuleCheck('rule', '%s:%s:%s' %
                                       (action, attr_name,
                                        sub_attr_name)) for
                      sub_attr_name in data if sub_attr_name in
                      target[attr_name]]
    return policy.AndCheck(sub_attr_rules)


def _build_list_of_subattrs_rule(attr_name, attribute_value, action):
    rules = []
    for sub_attr in attribute_value:
        if isinstance(sub_attr, dict):
            for k in sub_attr:
                rules.append(policy.RuleCheck(
                    'rule', '%s:%s:%s' % (action, attr_name, k)))
    if rules:
        return policy.AndCheck(rules)


def _process_rules_list(rules, match_rule):
    """Recursively walk a policy rule to extract a list of match entries."""
    if isinstance(match_rule, policy.RuleCheck):
        rules.append(match_rule.match)
    elif isinstance(match_rule, policy.AndCheck):
        for rule in match_rule.rules:
            _process_rules_list(rules, rule)
    return rules


def _build_match_rule(action, target, pluralized):
    """Create the rule to match for a given action.

    The policy rule to be matched is built in the following way:
    1) add entries for matching permission on objects
    2) add an entry for the specific action (e.g.: create_network)
    3) add an entry for attributes of a resource for which the action
       is being executed (e.g.: create_network:shared)
    4) add an entry for sub-attributes of a resource for which the
       action is being executed
       (e.g.: create_router:external_gateway_info:network_id)
    """
    match_rule = policy.RuleCheck('rule', action)
    registered_rule = _ENFORCER.registered_rules.get(action)
    if registered_rule and registered_rule.scope_types:
        match_rule.scope_types = registered_rule.scope_types

    resource, enforce_attr_based_check = get_resource_and_action(
        action, pluralized)
    if enforce_attr_based_check:
        # assigning to variable with short name for improving readability
{'fraction_non_alphanumeric': 0.07471058598878148, 'fraction_numerical': 0.005489915264351355, 'mean_word_length': 3.2150402414486923, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 8, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '5855017', 'n_tokens_mistral': 4795, 'n_tokens_neox': 4699, 'n_words': 1588}