Complete policy representation classes for extended permissions rules.

Related to #73.

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):

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_ALLOW: return "allowxperm"; break;
-            case QPOL_RULE_XPERMS_NEVERALLOW: return "neverallowx"; break;
+            case QPOL_RULE_XPERMS_NEVERALLOW: return "neverallowxperm"; break;
-            case QPOL_RULE_XPERMS_AUDITALLOW: return "auditallowx"; break;
+            case QPOL_RULE_XPERMS_AUDITALLOW: return "auditallowxperm"; break;
-            case QPOL_RULE_XPERMS_DONTAUDIT: return "dontauditx"; break;
+            case QPOL_RULE_XPERMS_DONTAUDIT: return "dontauditxperm"; break;
         }
     fail:
         return NULL;

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
 

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
-                # allow/dontaudit/auditallow/neverallow rules
+            perms = self.perms
-                perms = self.perms
+            if len(perms) > 1:
-                if len(perms) > 1:
+                self._rule_string += "{{ {0} }};".format(' '.join(perms))
-                    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
             else:
-                # extended avrules
+                # convert to list since sets cannot be indexed
-                xperms = self.xperms
+                self._rule_string += "{0};".format(list(perms)[0])
-                if len(xperms) > 1:
+
-                    self._rule_string += "{0} {{{1} }};".format(self.xperm_type, self.xperms_as_string)
+            try:
-                else:
+                self._rule_string += " [ {0.conditional} ]:{0.conditional_block}".format(self)
-                    self._rule_string += "{0} {1};".format(self.xperm_type, self.xperms_as_string)
+            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."""

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)