From c56e01bc8c25d3acff6eb6e783da11a7b84575d5 Mon Sep 17 00:00:00 2001 From: Chris PeBenito Date: Tue, 22 Mar 2016 10:26:43 -0400 Subject: [PATCH] Complete policy representation classes for extended permissions rules. Related to #73. --- setools/policyrep/__init__.py | 1 + setools/policyrep/qpol.i | 8 +- setools/policyrep/rule.py | 2 + setools/policyrep/terule.py | 161 ++++++++++++++++++++-------------- tests/policyrep/terule.py | 112 +++++++++++++++++++++++ 5 files changed, 213 insertions(+), 71 deletions(-) diff --git a/setools/policyrep/__init__.py b/setools/policyrep/__init__.py index 4f7b42b..7b23bc4 100644 --- a/setools/policyrep/__init__.py +++ b/setools/policyrep/__init__.py @@ -71,6 +71,7 @@ from . import xencontext # Classes useful for policyrep users: from . import exception from .netcontext import PortconProtocol, port_range +from .terule import ioctlSet class SELinuxPolicy(object): diff --git a/setools/policyrep/qpol.i b/setools/policyrep/qpol.i index cf02758..267250d 100644 --- a/setools/policyrep/qpol.i +++ b/setools/policyrep/qpol.i @@ -2680,10 +2680,10 @@ typedef struct qpol_avrule {} qpol_avrule_t; case QPOL_RULE_NEVERALLOW: return "neverallow"; break; case QPOL_RULE_AUDITALLOW: return "auditallow"; break; case QPOL_RULE_DONTAUDIT: return "dontaudit"; break; - case QPOL_RULE_XPERMS_ALLOW: return "allowx"; break; - case QPOL_RULE_XPERMS_NEVERALLOW: return "neverallowx"; break; - case QPOL_RULE_XPERMS_AUDITALLOW: return "auditallowx"; break; - case QPOL_RULE_XPERMS_DONTAUDIT: return "dontauditx"; break; + case QPOL_RULE_XPERMS_ALLOW: return "allowxperm"; break; + case QPOL_RULE_XPERMS_NEVERALLOW: return "neverallowxperm"; break; + case QPOL_RULE_XPERMS_AUDITALLOW: return "auditallowxperm"; break; + case QPOL_RULE_XPERMS_DONTAUDIT: return "dontauditxperm"; break; } fail: return NULL; diff --git a/setools/policyrep/rule.py b/setools/policyrep/rule.py index b7a6a9d..be6e8d0 100644 --- a/setools/policyrep/rule.py +++ b/setools/policyrep/rule.py @@ -25,6 +25,8 @@ class PolicyRule(symbol.PolicySymbol): """This is base class for policy rules.""" + extended = False + def __str__(self): raise NotImplementedError diff --git a/setools/policyrep/terule.py b/setools/policyrep/terule.py index e5bcfa4..cecf2a7 100644 --- a/setools/policyrep/terule.py +++ b/setools/policyrep/terule.py @@ -29,7 +29,10 @@ def te_rule_factory(policy, symbol): """Factory function for creating TE rule objects.""" if isinstance(symbol, qpol.qpol_avrule_t): - return AVRule(policy, symbol) + if symbol.is_extended(policy): + return AVRuleXperm(policy, symbol) + else: + return AVRule(policy, symbol) elif isinstance(symbol, (qpol.qpol_terule_t, qpol.qpol_filename_trans_t)): return TERule(policy, symbol) else: @@ -47,9 +50,11 @@ def expanded_te_rule_factory(original, source, target): if isinstance(original, AVRule): rule = ExpandedAVRule(original.policy, original.qpol_symbol) + elif isinstance(original, AVRuleXperm): + rule = ExpandedAVRuleXperm(original.policy, original.qpol_symbol) elif isinstance(original, TERule): rule = ExpandedTERule(original.policy, original.qpol_symbol) - elif isinstance(original, (ExpandedAVRule, ExpandedTERule)): + elif isinstance(original, (ExpandedAVRule, ExpandedAVRuleXperm, ExpandedTERule)): return original else: raise TypeError("The original rule must be a TE rule class.") @@ -64,7 +69,7 @@ def validate_ruletype(t): """Validate TE Rule types.""" if t not in ["allow", "auditallow", "dontaudit", "neverallow", "type_transition", "type_member", "type_change", - "allowx", "auditallowx", "dontauditx", "neverallowx"]: + "allowxperm", "auditallowxperm", "dontauditxperm", "neverallowxperm"]: raise exception.InvalidTERuleType("{0} is not a valid TE rule type.".format(t)) return t @@ -120,79 +125,26 @@ class AVRule(BaseTERule): except AttributeError: self._rule_string = "{0.ruletype} {0.source} {0.target}:{0.tclass} ".format(self) - if not self.qpol_symbol.is_extended(self.policy): - # allow/dontaudit/auditallow/neverallow rules - perms = self.perms - if len(perms) > 1: - self._rule_string += "{{ {0} }};".format(' '.join(perms)) - else: - # convert to list since sets cannot be indexed - self._rule_string += "{0};".format(list(perms)[0]) - - try: - self._rule_string += " [ {0.conditional} ]:{0.conditional_block}".format(self) - except exception.RuleNotConditional: - pass + # allow/dontaudit/auditallow/neverallow rules + perms = self.perms + if len(perms) > 1: + self._rule_string += "{{ {0} }};".format(' '.join(perms)) else: - # extended avrules - xperms = self.xperms - if len(xperms) > 1: - self._rule_string += "{0} {{{1} }};".format(self.xperm_type, self.xperms_as_string) - else: - self._rule_string += "{0} {1};".format(self.xperm_type, self.xperms_as_string) + # convert to list since sets cannot be indexed + self._rule_string += "{0};".format(list(perms)[0]) + + try: + self._rule_string += " [ {0.conditional} ]:{0.conditional_block}".format(self) + except exception.RuleNotConditional: + pass return self._rule_string @property def perms(self): """The rule's permission set.""" - if self.qpol_symbol.is_extended(self.policy): - raise exception.RuleUseError("{0} rules do not have permissions.".format(self.ruletype)) return set(self.qpol_symbol.perm_iter(self.policy)) - @property - def xperms_as_string(self): - """The rules extended permissions as a pretty string""" - if not self.qpol_symbol.is_extended(self.policy): - raise exception.RuleUseError("{0} rules do not have extended permissions.".format(self.ruletype)) - - def create_range_string(start, end): - if start == end: - return " 0x{0:04X}".format(start) - else: - return " 0x{0:04X}-0x{1:04X}".format(start, end) - - xperms_str = "" - xperms = self.qpol_symbol.xperm_iter(self.policy) - range_start = xperms.next() - range_end = range_start - - for xperm in xperms: - if xperm == range_end + 1: - range_end = xperm - else: - xperms_str += create_range_string(range_start, range_end) - range_start = xperm - range_end = xperm - - xperms_str += create_range_string(range_start, range_end) - - return xperms_str - - @property - def xperms(self): - """The rules extended permissions.""" - if not self.qpol_symbol.is_extended(self.policy): - raise exception.RuleUseError("{0} rules do not have extended permissions.".format(self.ruletype)) - return set(self.qpol_symbol.xperm_iter(self.policy)) - - @property - def xperm_type(self): - """The type of an extended permission.""" - if not self.qpol_symbol.is_extended(self.policy): - raise exception.RuleUseError("{0} rules do not have extended permissions.".format(self.ruletype)) - return self.qpol_symbol.xperm_type(self.policy) - @property def default(self): """The rule's default type.""" @@ -203,6 +155,72 @@ class AVRule(BaseTERule): raise exception.RuleUseError("{0} rules do not have file names".format(self.ruletype)) +class ioctlSet(set): + + """ + A set with an overridden str function which compresses + the output into ioctl ranges. + """ + + def __str__(self): + # generate short permission notation + perms = sorted(self) + shortlist = [] + for _, i in itertools.groupby(perms, key=lambda k, c=itertools.count(): k - next(c)): + group = list(i) + if len(group) > 1: + shortlist.append("0x{0:04x}-0x{1:04x}".format(group[0], group[-1])) + else: + shortlist.append("0x{0:04x}".format(group[0])) + + return " ".join(shortlist) + + def __repr__(self): + return "{{ {0} }}".format(self) + + def ranges(self): + """ + Return the number of ranges in the set. Main use + is to determine if brackets need to be used in + string output. + """ + return sum(1 for (_a, _b) in itertools.groupby( + sorted(self), key=lambda k, c=itertools.count(): k - next(c))) + + +class AVRuleXperm(AVRule): + + """An extended permission access vector type enforcement rule.""" + + extended = True + + def __str__(self): + try: + return self._rule_string + except AttributeError: + self._rule_string = "{0.ruletype} {0.source} {0.target}:{0.tclass} {0.xperm_type} ". \ + format(self) + + # generate short permission notation + perms = self.perms + if perms.ranges() > 1: + self._rule_string += "{{ {0} }};".format(perms) + else: + self._rule_string += "{0};".format(perms) + + return self._rule_string + + @property + def perms(self): + """The rule's extended permission set.""" + return ioctlSet(self.qpol_symbol.xperm_iter(self.policy)) + + @property + def xperm_type(self): + """The standard permission extended by these permissions (e.g. ioctl).""" + return self.qpol_symbol.xperm_type(self.policy) + + class TERule(BaseTERule): """A type_* type enforcement rule.""" @@ -276,6 +294,15 @@ class ExpandedAVRule(AVRule): origin = None +class ExpandedAVRuleXperm(AVRuleXperm): + + """An expanded extended permission access vector type enforcement rule.""" + + source = None + target = None + origin = None + + class ExpandedTERule(TERule): """An expanded type_* type enforcement rule.""" diff --git a/tests/policyrep/terule.py b/tests/policyrep/terule.py index dbfcd6f..235c965 100644 --- a/tests/policyrep/terule.py +++ b/tests/policyrep/terule.py @@ -37,6 +37,7 @@ class AVRuleTest(unittest.TestCase): def mock_avrule_factory(self, ruletype, source, target, tclass, perms, cond=None): mock_rule = Mock(qpol_avrule_t) + mock_rule.is_extended.return_value = False mock_rule.rule_type.return_value = ruletype mock_rule.source_type.return_value = source mock_rule.target_type.return_value = target @@ -151,6 +152,117 @@ class AVRuleTest(unittest.TestCase): ")") +@patch('setools.policyrep.boolcond.condexpr_factory', lambda x, y: y) +@patch('setools.policyrep.typeattr.type_or_attr_factory', lambda x, y: y) +@patch('setools.policyrep.objclass.class_factory', lambda x, y: y) +class AVRuleXpermTest(unittest.TestCase): + + def mock_avrule_factory(self, ruletype, source, target, tclass, xperm, perms): + mock_rule = Mock(qpol_avrule_t) + mock_rule.is_extended.return_value = True + mock_rule.rule_type.return_value = ruletype + mock_rule.source_type.return_value = source + mock_rule.target_type.return_value = target + mock_rule.object_class.return_value = tclass + mock_rule.xperm_type.return_value = xperm + mock_rule.xperm_iter = lambda x: iter(perms) + + # this actually comes out of condexpr_factory + # but it's simpler to have here + mock_rule.cond.side_effect = AttributeError + + return te_rule_factory(self.p, mock_rule) + + def setUp(self): + self.p = Mock(qpol_policy_t) + + def test_000_factory(self): + """AVRuleXperm factory lookup.""" + with self.assertRaises(TypeError): + te_rule_factory(self.p, "INVALID") + + def test_001_validate_ruletype(self): + """AVRuleXperm valid rule types.""" + for r in ["allowxperm", "neverallowxperm", "auditallowxperm", "dontauditxperm"]: + self.assertEqual(r, validate_ruletype(r)) + + def test_010_ruletype(self): + """AVRuleXperm rule type""" + rule = self.mock_avrule_factory("neverallowxperm", "a", "b", "c", "d", [0x0001]) + self.assertEqual("neverallowxperm", rule.ruletype) + + def test_020_source_type(self): + """AVRuleXperm source type""" + rule = self.mock_avrule_factory("allowxperm", "source20", "b", "c", "d", [0x0001]) + self.assertEqual("source20", rule.source) + + def test_030_target_type(self): + """AVRuleXperm target type""" + rule = self.mock_avrule_factory("allowxperm", "a", "target30", "c", "d", [0x0001]) + self.assertEqual("target30", rule.target) + + def test_040_object_class(self): + """AVRuleXperm object class""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "class40", "d", [0x0001]) + self.assertEqual("class40", rule.tclass) + + def test_050_permissions(self): + """AVRuleXperm permissions""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", [0x0001, 0x0002, 0x0003]) + self.assertSetEqual(set([0x0001, 0x0002, 0x0003]), rule.perms) + + def test_060_xperm_type(self): + """AVRuleXperm xperm type""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "xperm60", [0x0001]) + self.assertEqual("xperm60", rule.xperm_type) + + def test_070_unconditional(self): + """AVRuleXperm conditional expression (none)""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", [0x0001]) + with self.assertRaises(RuleNotConditional): + rule.conditional + + def test_080_default(self): + """AVRuleXperm default type""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", [0x0001]) + with self.assertRaises(RuleUseError): + rule.default + + def test_090_filename(self): + """AVRuleXperm filename (none)""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", [0x0001]) + with self.assertRaises(RuleUseError): + rule.filename + + def test_100_statement_one_perm(self): + """AVRuleXperm statement, one permission.""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", [0x0001]) + self.assertEqual("allowxperm a b:c d 0x0001;", rule.statement()) + + def test_101_statement_two_perms(self): + """AVRuleXperm statement, two permissions.""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", [0x0001, 0x0003]) + self.assertEqual(rule.statement(), "allowxperm a b:c d { 0x0001 0x0003 };") + + def test_102_statement_range_perms(self): + """AVRuleXperm statement, range of permissions.""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", + list(range(0x0010, 0x0015))) + self.assertEqual(rule.statement(), "allowxperm a b:c d 0x0010-0x0014;") + + def test_103_statement_single_perm_range_perms(self): + """AVRuleXperm statement, single perm with range of permissions.""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", + [0x0001, 0x0003, 0x0004, 0x0005]) + self.assertEqual(rule.statement(), "allowxperm a b:c d { 0x0001 0x0003-0x0005 };") + + def test_104_statement_two_range_perms(self): + """AVRuleXperm statement, two ranges of permissions.""" + rule = self.mock_avrule_factory("allowxperm", "a", "b", "c", "d", + [0x0003, 0x0004, 0x0005, 0x0007, 0x0008, 0x0009]) + self.assertEqual(rule.statement(), "allowxperm a b:c d { 0x0003-0x0005 0x0007-0x0009 };") + + @patch('setools.policyrep.boolcond.condexpr_factory', lambda x, y: y) @patch('setools.policyrep.typeattr.type_factory', lambda x, y: y) @patch('setools.policyrep.typeattr.type_or_attr_factory', lambda x, y: y)