Add optional exception for check_rules

Rather than checking for a True/False return from check_rules some
consumers may want it to raise an exception instead. This adds a
parameter, defaulting to False, which will cause check_rules to raise an
exception if set to True and a rule violation is found.

Co-Authored-By: ChangBo Guo(gcb) <eric.guo@easystack.cn>
Change-Id: Ie69bbc7081e84d019b85b62a45589ed331cc1f15

oslo_policy/policy.py

@@ -314,6 +314,13 @@ class PolicyNotRegistered(Exception):
         super(PolicyNotRegistered, self).__init__(msg)
 
 
+class InvalidDefinitionError(Exception):
+    def __init__(self, names):
+        msg = _('Policies %(names)s are not well defined. Check logs for '
+                'more details.') % {'names': names}
+        super(InvalidDefinitionError, self).__init__(msg)
+
+
 def parse_file_contents(data):
     """Parse the raw contents of a policy file.
 
@@ -541,7 +548,7 @@ class Enforcer(object):
             # Detect and log obvious incorrect rule definitions
             self.check_rules()
 
-    def check_rules(self):
+    def check_rules(self, raise_on_violation=False):
         """Look for rule definitions that are obviously incorrect."""
         undefined_checks = []
         cyclic_checks = []
@@ -561,6 +568,9 @@ class Enforcer(object):
             LOG.warning(_LW('Policies %(names)s are part of a cyclical '
                         'reference.'), {'names': cyclic_checks})
 
+        if raise_on_violation and violation:
+            raise InvalidDefinitionError(undefined_checks + cyclic_checks)
+
         return not violation
 
     def _undefined_check(self, check):

oslo_policy/tests/test_policy.py

@@ -841,7 +841,7 @@ class EnforcerCheckRulesTest(base.PolicyBaseTestCase):
     def test_no_violations(self):
         self.create_config_file('policy.json', POLICY_JSON_CONTENTS)
         self.enforcer.load_rules(True)
-        self.assertTrue(self.enforcer.check_rules())
+        self.assertTrue(self.enforcer.check_rules(raise_on_violation=True))
 
     def test_undefined_rule(self):
         rules = jsonutils.dumps({'foo': 'rule:bar'})
@@ -850,6 +850,14 @@ class EnforcerCheckRulesTest(base.PolicyBaseTestCase):
 
         self.assertFalse(self.enforcer.check_rules())
 
+    def test_undefined_rule_raises(self):
+        rules = jsonutils.dumps({'foo': 'rule:bar'})
+        self.create_config_file('policy.json', rules)
+        self.enforcer.load_rules(True)
+
+        self.assertRaises(policy.InvalidDefinitionError,
+                          self.enforcer.check_rules, raise_on_violation=True)
+
     def test_cyclical_rules(self):
         rules = jsonutils.dumps({'foo': 'rule:bar', 'bar': 'rule:foo'})
         self.create_config_file('policy.json', rules)
@@ -857,6 +865,14 @@ class EnforcerCheckRulesTest(base.PolicyBaseTestCase):
 
         self.assertFalse(self.enforcer.check_rules())
 
+    def test_cyclical_rules_raises(self):
+        rules = jsonutils.dumps({'foo': 'rule:bar', 'bar': 'rule:foo'})
+        self.create_config_file('policy.json', rules)
+        self.enforcer.load_rules(True)
+
+        self.assertRaises(policy.InvalidDefinitionError,
+                          self.enforcer.check_rules, raise_on_violation=True)
+
     def test_complex_cyclical_rules_false(self):
         rules = jsonutils.dumps({'foo': 'rule:bar',
                                  'bar': 'rule:baz and role:admin',