Merge utils: Improve rule priority support in is_covered/is_equal

- `is_covered` was not checking priorities when checking if a rule is
  covered. With this fix, a rule of lower priority can no longer cover a
  higher priority one.
- Fixes `is_equal(strict=False)` so that `priority=0` matches implicit
  priority (as it is defaulted to zero)
    
Signed-off-by: Maxime Bélair <maxime.belair@canonical.com>

MR: https://gitlab.com/apparmor/apparmor/-/merge_requests/1735
Approved-by: Georgia Garcia <georgia.garcia@canonical.com>
Approved-by: Christian Boltz <apparmor@cboltz.de>
Merged-by: Christian Boltz <apparmor@cboltz.de>

utils/apparmor/rule/__init__.py

@@ -176,7 +176,7 @@ class BaseRule(metaclass=ABCMeta):
         else:
             return self.get_clean(depth)
 
-    def is_covered(self, other_rule, check_allow_deny=True, check_audit=False):
+    def is_covered(self, other_rule, check_allow_deny=True, check_audit=False, check_priority=True):
         """check if other_rule is covered by this rule object"""
 
         if type(other_rule) is not type(self):
@@ -194,6 +194,9 @@ class BaseRule(metaclass=ABCMeta):
         if other_rule.audit and not self.audit:
             return False
 
+        if check_priority and (self.priority or 0) > (other_rule.priority or 0):
+            return False
+
         # still here? -> then the common part is covered, check rule-specific things now
         return self._is_covered_localvars(other_rule)
 
@@ -250,13 +253,14 @@ class BaseRule(metaclass=ABCMeta):
         """compare if rule_obj == self
            Calls _is_equal_localvars() to compare rule-specific variables"""
 
-        if (self.priority != rule_obj.priority
+        if ((self.priority or 0) != (rule_obj.priority or 0)
                 or self.audit != rule_obj.audit
                 or self.deny != rule_obj.deny):
             return False
 
         if strict and (
-            self.allow_keyword != rule_obj.allow_keyword
+            self.priority != rule_obj.priority
+            or self.allow_keyword != rule_obj.allow_keyword
             or self.comment != rule_obj.comment
             or self.raw_rule != rule_obj.raw_rule
         ):

utils/test/test-dbus.py

@@ -728,6 +728,17 @@ class DbusCoveredTest_11(DbusCoveredTest):
     )
 
 
+class DbusCoveredTest_Priority(DbusCoveredTest):
+    rule = 'dbus send,'
+
+    tests = (
+        #   rule                         equal  strict equal  covered  covered exact
+        ('priority=-1 dbus send,',       (False, False,       False,   False)),
+        ('priority=1 dbus send,',        (False, False,       True,    True)),
+        ('priority=0 dbus send,',        (True,  False,       True,    True)),
+    )
+
+
 class DbusCoveredTest_Invalid(AATest):
     def AASetup(self):
         #                        access               bus        path         name          interface    member        peername      peerlabel

utils/test/test-io_uring.py

@@ -179,6 +179,22 @@ class IOUringIsCoveredTest(AATest):
         self.assertFalse(obj.is_covered(IOUringRule(IOUringRule.ALL, 'foo')))
         self.assertFalse(obj.is_covered(IOUringRule(('sqpoll'), IOUringRule.ALL)))
 
+    def test_is_covered_priority(self):
+        obj = IOUringRule(IOUringRule.ALL, 'ba*', priority=0)
+        prio_obj = IOUringRule(IOUringRule.ALL, 'ba*', priority=1)
+        self.assertTrue(obj.is_covered(prio_obj))
+        self.assertFalse(prio_obj.is_covered(obj))
+
+    def test_is_covered_priority_2(self):
+        obj = IOUringRule(IOUringRule.ALL, 'ba*')
+        obj2 = IOUringRule(IOUringRule.ALL, 'ba*', priority=0)
+        self.assertTrue(obj.is_covered(obj2))
+        self.assertTrue(obj2.is_covered(obj))
+        self.assertTrue(obj.is_equal(obj2))
+        self.assertTrue(obj2.is_equal(obj))
+        self.assertFalse(obj.is_equal(obj2, strict=True))
+        self.assertFalse(obj2.is_equal(obj, strict=True))
+
 
 class IOUringLogprofHeaderTest(AATest):
     tests = (

utils/test/test-unix.py

@@ -143,6 +143,13 @@ class UnixIsCoveredTest(AATest):
             self.assertFalse(obj.is_covered(UnixRule(*test)), test)
             self.assertFalse(obj.is_equal(UnixRule(*test)))
 
+    def test_is_covered_priority(self):
+        obj = UnixRule(('accept', 'rw'), {'type': 'F*', 'protocol': 'AA'}, {'addr': 'AA'}, {'addr': 'AA', 'label': 'bb'}, priority=0)
+        prio_obj = UnixRule(('accept', 'rw'), {'type': 'F*', 'protocol': 'AA'}, {'addr': 'AA'}, {'addr': 'AA', 'label': 'bb'}, priority=1)
+
+        self.assertTrue(obj.is_covered(prio_obj))
+        self.assertFalse(prio_obj.is_covered(obj))
+
 
 class UnixLogprofHeaderTest(AATest):
     tests = (