# vim: tabstop=4 shiftwidth=4 softtabstop=4
-# Copyright (c) 2011 OpenStack, LLC.
+# Copyright (c) 2012 OpenStack, LLC.
# All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# License for the specific language governing permissions and limitations
# under the License.
-"""Common Policy Engine Implementation"""
+"""
+Common Policy Engine Implementation
+Policies can be expressed in one of two forms: A list of lists, or a
+string written in the new policy language.
+
+In the list-of-lists representation, each check inside the innermost
+list is combined as with an "and" conjunction--for that check to pass,
+all the specified checks must pass. These innermost lists are then
+combined as with an "or" conjunction. This is the original way of
+expressing policies, but there now exists a new way: the policy
+language.
+
+In the policy language, each check is specified the same way as in the
+list-of-lists representation: a simple "a:b" pair that is matched to
+the correct code to perform that check. However, conjunction
+operators are available, allowing for more expressiveness in crafting
+policies.
+
+As an example, take the following rule, expressed in the list-of-lists
+representation::
+
+ [["role:admin"], ["project_id:%(project_id)s", "role:projectadmin"]]
+
+In the policy language, this becomes::
+
+ role:admin or (project_id:%(project_id)s and role:projectadmin)
+
+The policy language also has the "not" operator, allowing a richer
+policy rule::
+
+ project_id:%(project_id)s and not role:dunce
+
+Finally, two special policy checks should be mentioned; the policy
+check "@" will always accept an access, and the policy check "!" will
+always reject an access. (Note that if a rule is either the empty
+list ("[]") or the empty string, this is equivalent to the "@" policy
+check.) Of these, the "!" policy check is probably the most useful,
+as it allows particular rules to be explicitly disabled.
+"""
+
+import abc
import logging
+import re
import urllib
+
import urllib2
from quantum.openstack.common.gettextutils import _
LOG = logging.getLogger(__name__)
-_BRAIN = None
+_rules = None
+_checks = {}
-def set_brain(brain):
- """Set the brain used by enforce().
+class Rules(dict):
+ """
+ A store for rules. Handles the default_rule setting directly.
+ """
- Defaults use Brain() if not set.
+ @classmethod
+ def load_json(cls, data, default_rule=None):
+ """
+ Allow loading of JSON rule data.
+ """
- """
- global _BRAIN
- _BRAIN = brain
+ # Suck in the JSON data and parse the rules
+ rules = dict((k, parse_rule(v)) for k, v in
+ jsonutils.loads(data).items())
+ return cls(rules, default_rule)
-def reset():
- """Clear the brain used by enforce()."""
- global _BRAIN
- _BRAIN = None
+ def __init__(self, rules=None, default_rule=None):
+ """Initialize the Rules store."""
+ super(Rules, self).__init__(rules or {})
+ self.default_rule = default_rule
-def enforce(match_list, target_dict, credentials_dict, exc=None,
- *args, **kwargs):
- """Enforces authorization of some rules against credentials.
+ def __missing__(self, key):
+ """Implements the default rule handling."""
- :param match_list: nested tuples of data to match against
+ # If the default rule isn't actually defined, do something
+ # reasonably intelligent
+ if not self.default_rule or self.default_rule not in self:
+ raise KeyError(key)
- The basic brain supports three types of match lists:
+ return self[self.default_rule]
- 1) rules
+ def __str__(self):
+ """Dumps a string representation of the rules."""
- looks like: ``('rule:compute:get_instance',)``
+ # Start by building the canonical strings for the rules
+ out_rules = {}
+ for key, value in self.items():
+ # Use empty string for singleton TrueCheck instances
+ if isinstance(value, TrueCheck):
+ out_rules[key] = ''
+ else:
+ out_rules[key] = str(value)
- Retrieves the named rule from the rules dict and recursively
- checks against the contents of the rule.
+ # Dump a pretty-printed JSON representation
+ return jsonutils.dumps(out_rules, indent=4)
- 2) roles
- looks like: ``('role:compute:admin',)``
+# Really have to figure out a way to deprecate this
+def set_rules(rules):
+ """Set the rules in use for policy checks."""
- Matches if the specified role is in credentials_dict['roles'].
+ global _rules
- 3) generic
+ _rules = rules
- looks like: ``('tenant_id:%(tenant_id)s',)``
- Substitutes values from the target dict into the match using
- the % operator and matches them against the creds dict.
+# Ditto
+def reset():
+ """Clear the rules used for policy checks."""
- Combining rules:
+ global _rules
- The brain returns True if any of the outer tuple of rules
- match and also True if all of the inner tuples match. You
- can use this to perform simple boolean logic. For
- example, the following rule would return True if the creds
- contain the role 'admin' OR the if the tenant_id matches
- the target dict AND the the creds contains the role
- 'compute_sysadmin':
+ _rules = None
- ::
- {
- "rule:combined": (
- 'role:admin',
- ('tenant_id:%(tenant_id)s', 'role:compute_sysadmin')
- )
- }
+def check(rule, target, creds, exc=None, *args, **kwargs):
+ """
+ Checks authorization of a rule against the target and credentials.
+
+ :param rule: The rule to evaluate.
+ :param target: As much information about the object being operated
+ on as possible, as a dictionary.
+ :param creds: As much information about the user performing the
+ action as possible, as a dictionary.
+ :param exc: Class of the exception to raise if the check fails.
+ Any remaining arguments passed to check() (both
+ positional and keyword arguments) will be passed to
+ the exception class. If exc is not provided, returns
+ False.
+
+ :return: Returns False if the policy does not allow the action and
+ exc is not provided; otherwise, returns a value that
+ evaluates to True. Note: for rules using the "case"
+ expression, this True value will be the specified string
+ from the expression.
+ """
- Note that rule and role are reserved words in the credentials match, so
- you can't match against properties with those names. Custom brains may
- also add new reserved words. For example, the HttpBrain adds http as a
- reserved word.
+ # Allow the rule to be a Check tree
+ if isinstance(rule, BaseCheck):
+ result = rule(target, creds)
+ elif not _rules:
+ # No rules to reference means we're going to fail closed
+ result = False
+ else:
+ try:
+ # Evaluate the rule
+ result = _rules[rule](target, creds)
+ except KeyError:
+ # If the rule doesn't exist, fail closed
+ result = False
- :param target_dict: dict of object properties
+ # If it is False, raise the exception if requested
+ if exc and result is False:
+ raise exc(*args, **kwargs)
- Target dicts contain as much information as we can about the object being
- operated on.
+ return result
- :param credentials_dict: dict of actor properties
- Credentials dicts contain as much information as we can about the user
- performing the action.
+class BaseCheck(object):
+ """
+ Abstract base class for Check classes.
+ """
- :param exc: exception to raise
+ __metaclass__ = abc.ABCMeta
+
+ @abc.abstractmethod
+ def __str__(self):
+ """
+ Retrieve a string representation of the Check tree rooted at
+ this node.
+ """
+
+ pass
+
+ @abc.abstractmethod
+ def __call__(self, target, cred):
+ """
+ Perform the check. Returns False to reject the access or a
+ true value (not necessary True) to accept the access.
+ """
+
+ pass
- Class of the exception to raise if the check fails. Any remaining
- arguments passed to enforce() (both positional and keyword arguments)
- will be passed to the exception class. If exc is not provided, returns
- False.
- :return: True if the policy allows the action
- :return: False if the policy does not allow the action and exc is not set
+class FalseCheck(BaseCheck):
"""
- global _BRAIN
- if not _BRAIN:
- _BRAIN = Brain()
- if not _BRAIN.check(match_list, target_dict, credentials_dict):
- if exc:
- raise exc(*args, **kwargs)
+ A policy check that always returns False (disallow).
+ """
+
+ def __str__(self):
+ """Return a string representation of this check."""
+
+ return "!"
+
+ def __call__(self, target, cred):
+ """Check the policy."""
+
return False
- return True
-class Brain(object):
- """Implements policy checking."""
+class TrueCheck(BaseCheck):
+ """
+ A policy check that always returns True (allow).
+ """
- _checks = {}
+ def __str__(self):
+ """Return a string representation of this check."""
- @classmethod
- def _register(cls, name, func):
- cls._checks[name] = func
+ return "@"
- @classmethod
- def load_json(cls, data, default_rule=None):
- """Init a brain using json instead of a rules dictionary."""
- rules_dict = jsonutils.loads(data)
- return cls(rules=rules_dict, default_rule=default_rule)
+ def __call__(self, target, cred):
+ """Check the policy."""
- def __init__(self, rules=None, default_rule=None):
- if self.__class__ != Brain:
- LOG.warning(_("Inheritance-based rules are deprecated; use "
- "the default brain instead of %s.") %
- self.__class__.__name__)
+ return True
- self.rules = rules or {}
- self.default_rule = default_rule
- def add_rule(self, key, match):
- self.rules[key] = match
+class Check(BaseCheck):
+ """
+ A base class to allow for user-defined policy checks.
+ """
- def _check(self, match, target_dict, cred_dict):
- try:
- match_kind, match_value = match.split(':', 1)
- except Exception:
- LOG.exception(_("Failed to understand rule %(match)r") % locals())
- # If the rule is invalid, fail closed
- return False
+ def __init__(self, kind, match):
+ """
+ :param kind: The kind of the check, i.e., the field before the
+ ':'.
+ :param match: The match of the check, i.e., the field after
+ the ':'.
+ """
- func = None
- try:
- old_func = getattr(self, '_check_%s' % match_kind)
- except AttributeError:
- func = self._checks.get(match_kind, self._checks.get(None, None))
- else:
- LOG.warning(_("Inheritance-based rules are deprecated; update "
- "_check_%s") % match_kind)
- func = (lambda brain, kind, value, target, cred:
- old_func(value, target, cred))
-
- if not func:
- LOG.error(_("No handler for matches of kind %s") % match_kind)
- # Fail closed
- return False
+ self.kind = kind
+ self.match = match
+
+ def __str__(self):
+ """Return a string representation of this check."""
+
+ return "%s:%s" % (self.kind, self.match)
+
+
+class NotCheck(BaseCheck):
+ """
+ A policy check that inverts the result of another policy check.
+ Implements the "not" operator.
+ """
+
+ def __init__(self, rule):
+ """
+ Initialize the 'not' check.
+
+ :param rule: The rule to negate. Must be a Check.
+ """
+
+ self.rule = rule
+
+ def __str__(self):
+ """Return a string representation of this check."""
+
+ return "not %s" % self.rule
- return func(self, match_kind, match_value, target_dict, cred_dict)
+ def __call__(self, target, cred):
+ """
+ Check the policy. Returns the logical inverse of the wrapped
+ check.
+ """
+
+ return not self.rule(target, cred)
+
+
+class AndCheck(BaseCheck):
+ """
+ A policy check that requires that a list of other checks all
+ return True. Implements the "and" operator.
+ """
- def check(self, match_list, target_dict, cred_dict):
- """Checks authorization of some rules against credentials.
+ def __init__(self, rules):
+ """
+ Initialize the 'and' check.
+
+ :param rules: A list of rules that will be tested.
+ """
- Detailed description of the check with examples in policy.enforce().
+ self.rules = rules
- :param match_list: nested tuples of data to match against
- :param target_dict: dict of object properties
- :param credentials_dict: dict of actor properties
+ def __str__(self):
+ """Return a string representation of this check."""
- :returns: True if the check passes
+ return "(%s)" % ' and '.join(str(r) for r in self.rules)
+ def __call__(self, target, cred):
+ """
+ Check the policy. Requires that all rules accept in order to
+ return True.
"""
- if not match_list:
- return True
- for and_list in match_list:
- if isinstance(and_list, basestring):
- and_list = (and_list,)
- if all([self._check(item, target_dict, cred_dict)
- for item in and_list]):
+
+ for rule in self.rules:
+ if not rule(target, cred):
+ return False
+
+ return True
+
+ def add_check(self, rule):
+ """
+ Allows addition of another rule to the list of rules that will
+ be tested. Returns the AndCheck object for convenience.
+ """
+
+ self.rules.append(rule)
+ return self
+
+
+class OrCheck(BaseCheck):
+ """
+ A policy check that requires that at least one of a list of other
+ checks returns True. Implements the "or" operator.
+ """
+
+ def __init__(self, rules):
+ """
+ Initialize the 'or' check.
+
+ :param rules: A list of rules that will be tested.
+ """
+
+ self.rules = rules
+
+ def __str__(self):
+ """Return a string representation of this check."""
+
+ return "(%s)" % ' or '.join(str(r) for r in self.rules)
+
+ def __call__(self, target, cred):
+ """
+ Check the policy. Requires that at least one rule accept in
+ order to return True.
+ """
+
+ for rule in self.rules:
+ if rule(target, cred):
return True
+
return False
+ def add_check(self, rule):
+ """
+ Allows addition of another rule to the list of rules that will
+ be tested. Returns the OrCheck object for convenience.
+ """
-class HttpBrain(Brain):
- """A brain that can check external urls for policy.
+ self.rules.append(rule)
+ return self
- Posts json blobs for target and credentials.
- Note that this brain is deprecated; the http check is registered
- by default.
+def _parse_check(rule):
+ """
+ Parse a single base check rule into an appropriate Check object.
+ """
+
+ # Handle the special checks
+ if rule == '!':
+ return FalseCheck()
+ elif rule == '@':
+ return TrueCheck()
+
+ try:
+ kind, match = rule.split(':', 1)
+ except Exception:
+ LOG.exception(_("Failed to understand rule %(rule)s") % locals())
+ # If the rule is invalid, we'll fail closed
+ return FalseCheck()
+
+ # Find what implements the check
+ if kind in _checks:
+ return _checks[kind](kind, match)
+ elif None in _checks:
+ return _checks[None](kind, match)
+ else:
+ LOG.error(_("No handler for matches of kind %s") % kind)
+ return FalseCheck()
+
+
+def _parse_list_rule(rule):
+ """
+ Provided for backwards compatibility. Translates the old
+ list-of-lists syntax into a tree of Check objects.
"""
- pass
+ # Empty rule defaults to True
+ if not rule:
+ return TrueCheck()
+
+ # Outer list is joined by "or"; inner list by "and"
+ or_list = []
+ for inner_rule in rule:
+ # Elide empty inner lists
+ if not inner_rule:
+ continue
+
+ # Handle bare strings
+ if isinstance(inner_rule, basestring):
+ inner_rule = [inner_rule]
+
+ # Parse the inner rules into Check objects
+ and_list = [_parse_check(r) for r in inner_rule]
+
+ # Append the appropriate check to the or_list
+ if len(and_list) == 1:
+ or_list.append(and_list[0])
+ else:
+ or_list.append(AndCheck(and_list))
+
+ # If we have only one check, omit the "or"
+ if len(or_list) == 0:
+ return FalseCheck()
+ elif len(or_list) == 1:
+ return or_list[0]
+
+ return OrCheck(or_list)
+
+
+# Used for tokenizing the policy language
+_tokenize_re = re.compile(r'\s+')
+
+
+def _parse_tokenize(rule):
+ """
+ Tokenizer for the policy language.
+
+ Most of the single-character tokens are specified in the
+ _tokenize_re; however, parentheses need to be handled specially,
+ because they can appear inside a check string. Thankfully, those
+ parentheses that appear inside a check string can never occur at
+ the very beginning or end ("%(variable)s" is the correct syntax).
+ """
+
+ for tok in _tokenize_re.split(rule):
+ # Skip empty tokens
+ if not tok or tok.isspace():
+ continue
+
+ # Handle leading parens on the token
+ clean = tok.lstrip('(')
+ for i in range(len(tok) - len(clean)):
+ yield '(', '('
+
+ # If it was only parentheses, continue
+ if not clean:
+ continue
+ else:
+ tok = clean
+
+ # Handle trailing parens on the token
+ clean = tok.rstrip(')')
+ trail = len(tok) - len(clean)
+
+ # Yield the cleaned token
+ lowered = clean.lower()
+ if lowered in ('and', 'or', 'not'):
+ # Special tokens
+ yield lowered, clean
+ elif clean:
+ # Not a special token, but not composed solely of ')'
+ if len(tok) >= 2 and ((tok[0], tok[-1]) in
+ [('"', '"'), ("'", "'")]):
+ # It's a quoted string
+ yield 'string', tok[1:-1]
+ else:
+ yield 'check', _parse_check(clean)
+
+ # Yield the trailing parens
+ for i in range(trail):
+ yield ')', ')'
+
+
+class ParseStateMeta(type):
+ """
+ Metaclass for the ParseState class. Facilitates identifying
+ reduction methods.
+ """
+
+ def __new__(mcs, name, bases, cls_dict):
+ """
+ Create the class. Injects the 'reducers' list, a list of
+ tuples matching token sequences to the names of the
+ corresponding reduction methods.
+ """
+
+ reducers = []
+
+ for key, value in cls_dict.items():
+ if not hasattr(value, 'reducers'):
+ continue
+ for reduction in value.reducers:
+ reducers.append((reduction, key))
+
+ cls_dict['reducers'] = reducers
+
+ return super(ParseStateMeta, mcs).__new__(mcs, name, bases, cls_dict)
+
+
+def reducer(*tokens):
+ """
+ Decorator for reduction methods. Arguments are a sequence of
+ tokens, in order, which should trigger running this reduction
+ method.
+ """
+
+ def decorator(func):
+ # Make sure we have a list of reducer sequences
+ if not hasattr(func, 'reducers'):
+ func.reducers = []
+
+ # Add the tokens to the list of reducer sequences
+ func.reducers.append(list(tokens))
+
+ return func
+
+ return decorator
+
+
+class ParseState(object):
+ """
+ Implement the core of parsing the policy language. Uses a greedy
+ reduction algorithm to reduce a sequence of tokens into a single
+ terminal, the value of which will be the root of the Check tree.
+
+ Note: error reporting is rather lacking. The best we can get with
+ this parser formulation is an overall "parse failed" error.
+ Fortunately, the policy language is simple enough that this
+ shouldn't be that big a problem.
+ """
+
+ __metaclass__ = ParseStateMeta
+
+ def __init__(self):
+ """Initialize the ParseState."""
+
+ self.tokens = []
+ self.values = []
+
+ def reduce(self):
+ """
+ Perform a greedy reduction of the token stream. If a reducer
+ method matches, it will be executed, then the reduce() method
+ will be called recursively to search for any more possible
+ reductions.
+ """
+
+ for reduction, methname in self.reducers:
+ if (len(self.tokens) >= len(reduction) and
+ self.tokens[-len(reduction):] == reduction):
+ # Get the reduction method
+ meth = getattr(self, methname)
+
+ # Reduce the token stream
+ results = meth(*self.values[-len(reduction):])
+
+ # Update the tokens and values
+ self.tokens[-len(reduction):] = [r[0] for r in results]
+ self.values[-len(reduction):] = [r[1] for r in results]
+
+ # Check for any more reductions
+ return self.reduce()
+
+ def shift(self, tok, value):
+ """Adds one more token to the state. Calls reduce()."""
+
+ self.tokens.append(tok)
+ self.values.append(value)
+
+ # Do a greedy reduce...
+ self.reduce()
+
+ @property
+ def result(self):
+ """
+ Obtain the final result of the parse. Raises ValueError if
+ the parse failed to reduce to a single result.
+ """
+
+ if len(self.values) != 1:
+ raise ValueError("Could not parse rule")
+ return self.values[0]
+
+ @reducer('(', 'check', ')')
+ @reducer('(', 'and_expr', ')')
+ @reducer('(', 'or_expr', ')')
+ def _wrap_check(self, _p1, check, _p2):
+ """Turn parenthesized expressions into a 'check' token."""
+
+ return [('check', check)]
+
+ @reducer('check', 'and', 'check')
+ def _make_and_expr(self, check1, _and, check2):
+ """
+ Create an 'and_expr' from two checks joined by the 'and'
+ operator.
+ """
+
+ return [('and_expr', AndCheck([check1, check2]))]
+
+ @reducer('and_expr', 'and', 'check')
+ def _extend_and_expr(self, and_expr, _and, check):
+ """
+ Extend an 'and_expr' by adding one more check.
+ """
+
+ return [('and_expr', and_expr.add_check(check))]
+
+ @reducer('check', 'or', 'check')
+ def _make_or_expr(self, check1, _or, check2):
+ """
+ Create an 'or_expr' from two checks joined by the 'or'
+ operator.
+ """
+
+ return [('or_expr', OrCheck([check1, check2]))]
+
+ @reducer('or_expr', 'or', 'check')
+ def _extend_or_expr(self, or_expr, _or, check):
+ """
+ Extend an 'or_expr' by adding one more check.
+ """
+
+ return [('or_expr', or_expr.add_check(check))]
+
+ @reducer('not', 'check')
+ def _make_not_expr(self, _not, check):
+ """Invert the result of another check."""
+
+ return [('check', NotCheck(check))]
+
+
+def _parse_text_rule(rule):
+ """
+ Translates a policy written in the policy language into a tree of
+ Check objects.
+ """
+
+ # Empty rule means always accept
+ if not rule:
+ return TrueCheck()
+
+ # Parse the token stream
+ state = ParseState()
+ for tok, value in _parse_tokenize(rule):
+ state.shift(tok, value)
+
+ try:
+ return state.result
+ except ValueError:
+ # Couldn't parse the rule
+ LOG.exception(_("Failed to understand rule %(rule)r") % locals())
+
+ # Fail closed
+ return FalseCheck()
+
+
+def parse_rule(rule):
+ """
+ Parses a policy rule into a tree of Check objects.
+ """
+
+ # If the rule is a string, it's in the policy language
+ if isinstance(rule, basestring):
+ return _parse_text_rule(rule)
+ return _parse_list_rule(rule)
def register(name, func=None):
"""
- Register a function as a policy check.
+ Register a function or Check class as a policy check.
:param name: Gives the name of the check type, e.g., 'rule',
- 'role', etc. If name is None, a default function
+ 'role', etc. If name is None, a default check type
will be registered.
- :param func: If given, provides the function to register. If not
- given, returns a function taking one argument to
- specify the function to register, allowing use as a
- decorator.
+ :param func: If given, provides the function or class to register.
+ If not given, returns a function taking one argument
+ to specify the function or class to register,
+ allowing use as a decorator.
"""
- # Perform the actual decoration by registering the function.
- # Returns the function for compliance with the decorator
- # interface.
+ # Perform the actual decoration by registering the function or
+ # class. Returns the function or class for compliance with the
+ # decorator interface.
def decorator(func):
- # Register the function
- Brain._register(name, func)
+ _checks[name] = func
return func
- # If the function is given, do the registration
+ # If the function or class is given, do the registration
if func:
return decorator(func)
@register("rule")
-def _check_rule(brain, match_kind, match, target_dict, cred_dict):
- """Recursively checks credentials based on the brains rules."""
- try:
- new_match_list = brain.rules[match]
- except KeyError:
- if brain.default_rule and match != brain.default_rule:
- new_match_list = ('rule:%s' % brain.default_rule,)
- else:
- return False
+class RuleCheck(Check):
+ def __call__(self, target, creds):
+ """
+ Recursively checks credentials based on the defined rules.
+ """
- return brain.check(new_match_list, target_dict, cred_dict)
+ try:
+ return _rules[self.match](target, creds)
+ except KeyError:
+ # We don't have any matching rule; fail closed
+ return False
@register("role")
-def _check_role(brain, match_kind, match, target_dict, cred_dict):
- """Check that there is a matching role in the cred dict."""
- return match.lower() in [x.lower() for x in cred_dict['roles']]
+class RoleCheck(Check):
+ def __call__(self, target, creds):
+ """Check that there is a matching role in the cred dict."""
+
+ return self.match.lower() in [x.lower() for x in creds['roles']]
@register('http')
-def _check_http(brain, match_kind, match, target_dict, cred_dict):
- """Check http: rules by calling to a remote server.
+class HttpCheck(Check):
+ def __call__(self, target, creds):
+ """
+ Check http: rules by calling to a remote server.
- This example implementation simply verifies that the response is
- exactly 'True'. A custom brain using response codes could easily
- be implemented.
+ This example implementation simply verifies that the response
+ is exactly 'True'.
+ """
- """
- url = 'http:' + (match % target_dict)
- data = {'target': jsonutils.dumps(target_dict),
- 'credentials': jsonutils.dumps(cred_dict)}
- post_data = urllib.urlencode(data)
- f = urllib2.urlopen(url, post_data)
- return f.read() == "True"
+ url = ('http:' + self.match) % target
+ data = {'target': jsonutils.dumps(target),
+ 'credentials': jsonutils.dumps(creds)}
+ post_data = urllib.urlencode(data)
+ f = urllib2.urlopen(url, post_data)
+ return f.read() == "True"
@register(None)
-def _check_generic(brain, match_kind, match, target_dict, cred_dict):
- """Check an individual match.
-
- Matches look like:
+class GenericCheck(Check):
+ def __call__(self, target, creds):
+ """
+ Check an individual match.
- tenant:%(tenant_id)s
- role:compute:admin
+ Matches look like:
- """
+ tenant:%(tenant_id)s
+ role:compute:admin
+ """
- # TODO(termie): do dict inspection via dot syntax
- match = match % target_dict
- if match_kind in cred_dict:
- return match == unicode(cred_dict[match_kind])
- return False
+ # TODO(termie): do dict inspection via dot syntax
+ match = self.match % target
+ if self.kind in creds:
+ return match == unicode(creds[self.kind])
+ return False
# pass _set_brain to read_cached_file so that the policy brain
# is reset only if the file has changed
utils.read_cached_file(_POLICY_PATH, _POLICY_CACHE,
- reload_func=_set_brain)
+ reload_func=_set_rules)
def get_resource_and_action(action):
return ("%ss" % data[-1], data[0] != 'get')
-def _set_brain(data):
+def _set_rules(data):
default_rule = 'default'
- policy.set_brain(policy.Brain.load_json(data, default_rule))
+ policy.set_rules(policy.Rules.load_json(data, default_rule))
def _is_attribute_explicitly_set(attribute_name, resource, target):
return target
-def _build_match_list(action, target):
- """Create the list of rules to match for a given action.
+def _build_match_rule(action, target):
+ """Create the rule to match for a given action.
- The list of policy rules to be matched is built in the following way:
+ 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
"""
- match_list = ('rule:%s' % action,)
+ match_rule = policy.RuleCheck('rule', action)
resource, is_write = get_resource_and_action(action)
if is_write:
# assigning to variable with short name for improving readability
target):
attribute = res_map[resource][attribute_name]
if 'enforce_policy' in attribute and is_write:
- match_list += ('rule:%s:%s' % (action,
- attribute_name),)
- return [match_list]
+ attr_rule = policy.RuleCheck('rule', '%s:%s' %
+ (action, attribute_name))
+ match_rule = policy.AndCheck([match_rule, attr_rule])
+
+ return match_rule
@policy.register('field')
-def check_field(brain, match_kind, match, target_dict, cred_dict):
- # If this method is invoked for the wrong kind of match
- # which should never happen, just skip the check and don't
- # fail the policy evaluation
- if match_kind != 'field':
- LOG.warning("Field check function invoked with wrong match_kind:%s",
- match_kind)
- return True
- resource, field_value = match.split(':', 1)
- field, value = field_value.split('=', 1)
- target_value = target_dict.get(field)
- # target_value might be a boolean, explicitly compare with None
- if target_value is None:
- LOG.debug("Unable to find requested field: %s in target: %s",
- field, target_dict)
- return False
- # Value migth need conversion - we need help from the attribute map
- conv_func = attributes.RESOURCE_ATTRIBUTE_MAP[resource][field].get(
- 'convert_to', lambda x: x)
- if target_value != conv_func(value):
- LOG.debug("%s does not match the value in the target object:%s",
- conv_func(value), target_value)
- return False
- # If we manage to get here, the policy check is successful
- return True
+class FieldCheck(policy.Check):
+ def __init__(self, kind, match):
+ # Process the match
+ resource, field_value = match.split(':', 1)
+ field, value = field_value.split('=', 1)
+
+ super(FieldCheck, self).__init__(kind, '%s:%s:%s' %
+ (resource, field, value))
+
+ # Value might need conversion - we need help from the attribute map
+ try:
+ attr = attributes.RESOURCE_ATTRIBUTE_MAP[resource][field]
+ conv_func = attr['convert_to']
+ except KeyError:
+ conv_func = lambda x: x
+
+ self.field = field
+ self.value = conv_func(value)
+
+ def __call__(self, target_dict, cred_dict):
+ target_value = target_dict.get(self.field)
+ # target_value might be a boolean, explicitly compare with None
+ if target_value is None:
+ LOG.debug("Unable to find requested field: %s in target: %s",
+ self.field, target_dict)
+ return False
+
+ return target_value == self.value
def check(context, action, target, plugin=None):
"""
init()
real_target = _build_target(action, target, plugin, context)
- match_list = _build_match_list(action, real_target)
+ match_rule = _build_match_rule(action, real_target)
credentials = context.to_dict()
- return policy.enforce(match_list, real_target, credentials)
+ return policy.check(match_rule, real_target, credentials)
def enforce(context, action, target, plugin=None):
init()
real_target = _build_target(action, target, plugin, context)
- match_list = _build_match_list(action, real_target)
+ match_rule = _build_match_rule(action, real_target)
credentials = context.to_dict()
- policy.enforce(match_list, real_target, credentials,
- exceptions.PolicyNotAuthorized, action=action)
+ return policy.check(match_rule, real_target, credentials,
+ exceptions.PolicyNotAuthorized, action=action)
tmpfilename = os.path.join(tmpdir, 'policy')
action = "example:test"
with open(tmpfilename, "w") as policyfile:
- policyfile.write("""{"example:test": []}""")
+ policyfile.write("""{"example:test": ""}""")
policy.enforce(self.context, action, self.target)
with open(tmpfilename, "w") as policyfile:
- policyfile.write("""{"example:test": ["false:false"]}""")
+ policyfile.write("""{"example:test": "!"}""")
# NOTE(vish): reset stored policy cache so we don't have to
# sleep(1)
policy._POLICY_CACHE = {}
# NOTE(vish): preload rules to circumvent reloading from file
policy.init()
rules = {
- "true": [],
- "example:allowed": [],
- "example:denied": [["false:false"]],
- "example:get_http": [["http:http://www.example.com"]],
- "example:my_file": [["role:compute_admin"],
- ["tenant_id:%(tenant_id)s"]],
- "example:early_and_fail": [["false:false", "rule:true"]],
- "example:early_or_success": [["rule:true"], ["false:false"]],
- "example:lowercase_admin": [["role:admin"], ["role:sysadmin"]],
- "example:uppercase_admin": [["role:ADMIN"], ["role:sysadmin"]],
+ "true": '@',
+ "example:allowed": '@',
+ "example:denied": '!',
+ "example:get_http": "http:http://www.example.com",
+ "example:my_file": "role:compute_admin or tenant_id:%(tenant_id)s",
+ "example:early_and_fail": "! and @",
+ "example:early_or_success": "@ or !",
+ "example:lowercase_admin": "role:admin or role:sysadmin",
+ "example:uppercase_admin": "role:ADMIN or role:sysadmin",
}
- # NOTE(vish): then overload underlying brain
- common_policy.set_brain(common_policy.HttpBrain(rules))
+ # NOTE(vish): then overload underlying rules
+ common_policy.set_rules(common_policy.Rules(
+ dict((k, common_policy.parse_rule(v))
+ for k, v in rules.items())))
self.context = context.Context('fake', 'fake', roles=['member'])
self.target = {}
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
self.context, action, self.target)
+ def test_check_bad_action_noraise(self):
+ action = "example:denied"
+ result = policy.check(self.context, action, self.target)
+ self.assertEqual(result, False)
+
def test_enforce_good_action(self):
action = "example:allowed"
- policy.enforce(self.context, action, self.target)
+ result = policy.enforce(self.context, action, self.target)
+ self.assertEqual(result, True)
def test_enforce_http_true(self):
action = "example:get_http"
target = {}
result = policy.enforce(self.context, action, target)
- self.assertEqual(result, None)
+ self.assertEqual(result, True)
def test_enforce_http_false(self):
policy.init()
self.rules = {
- "default": [],
- "example:exist": [["false:false"]]
+ "default": '',
+ "example:exist": '!',
}
- self._set_brain('default')
+ self._set_rules('default')
self.context = context.Context('fake', 'fake')
- def _set_brain(self, default_rule):
- brain = common_policy.HttpBrain(self.rules, default_rule)
- common_policy.set_brain(brain)
+ def _set_rules(self, default_rule):
+ rules = common_policy.Rules(
+ dict((k, common_policy.parse_rule(v))
+ for k, v in self.rules.items()), default_rule)
+ common_policy.set_rules(rules)
def tearDown(self):
super(DefaultPolicyTestCase, self).tearDown()
policy.enforce(self.context, "example:noexist", {})
def test_default_not_found(self):
- self._set_brain("default_noexist")
+ self._set_rules("default_noexist")
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
self.context, "example:noexist", {})
super(QuantumPolicyTestCase, self).setUp()
policy.reset()
policy.init()
- self.rules = {
- "admin_or_network_owner": [["role:admin"],
- ["tenant_id:%(network_tenant_id)s"]],
- "admin_or_owner": [["role:admin"], ["tenant_id:%(tenant_id)s"]],
- "admin_only": [["role:admin"]],
- "regular_user": [["role:user"]],
- "shared": [["field:networks:shared=True"]],
- "external": [["field:networks:router:external=True"]],
- "default": [],
-
- "create_network": [["rule:admin_or_owner"]],
- "create_network:shared": [["rule:admin_only"]],
- "update_network": [],
- "update_network:shared": [["rule:admin_only"]],
-
- "get_network": [["rule:admin_or_owner"],
- ["rule:shared"],
- ["rule:external"]],
- "create_port:mac": [["rule:admin_or_network_owner"]],
- }
+ self.rules = dict((k, common_policy.parse_rule(v)) for k, v in {
+ "admin_or_network_owner": "role:admin or "
+ "tenant_id:%(network_tenant_id)s",
+ "admin_or_owner": "role:admin or tenant_id:%(tenant_id)s",
+ "admin_only": "role:admin",
+ "regular_user": "role:user",
+ "shared": "field:networks:shared=True",
+ "external": "field:networks:router:external=True",
+ "default": '@',
+
+ "create_network": "rule:admin_or_owner",
+ "create_network:shared": "rule:admin_only",
+ "update_network": '@',
+ "update_network:shared": "rule:admin_only",
+
+ "get_network": "rule:admin_or_owner or "
+ "rule:shared or "
+ "rule:external",
+ "create_port:mac": "rule:admin_or_network_owner",
+ }.items())
def fakepolicyinit():
- common_policy.set_brain(common_policy.Brain(self.rules))
+ common_policy.set_rules(common_policy.Rules(self.rules))
self.patcher = mock.patch.object(quantum.policy,
'init',
context, action, target, None)
else:
result = policy.enforce(context, action, target, None)
- self.assertEqual(result, None)
+ self.assertEqual(result, True)
def _test_nonadmin_action_on_attr(self, action, attr, value,
exception=None):
admin_context = context.get_admin_context()
target = {'shared': True}
result = policy.enforce(admin_context, action, target, None)
- self.assertEqual(result, None)
+ self.assertEqual(result, True)
def test_enforce_adminonly_attribute_create(self):
self._test_enforce_adminonly_attribute('create_network')
def test_enforce_adminonly_attribute_update(self):
self._test_enforce_adminonly_attribute('update_network')
- def test_enforce_adminoly_attribute_nonadminctx_returns_403(self):
+ def test_enforce_adminonly_attribute_nonadminctx_returns_403(self):
action = "create_network"
target = {'shared': True, 'tenant_id': 'somebody_else'}
self.assertRaises(exceptions.PolicyNotAuthorized, policy.enforce,
action = "get_network"
target = {'shared': True, 'tenant_id': 'somebody_else'}
result = policy.enforce(self.context, action, target, None)
- self.assertIsNone(result)
+ self.assertTrue(result)
def test_enforce_parentresource_owner(self):
with mock.patch.object(self.plugin, 'get_network', new=fakegetnetwork):
target = {'network_id': 'whatever'}
result = policy.enforce(self.context, action, target, self.plugin)
- self.assertIsNone(result)
+ self.assertTrue(result)
Code Review / openstack-build / neutron-build.git / commitdiff