From ab9d359405df0512d2af3ab9c0c3c5bdd5071245 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maxime=20B=C3=A9lair?= Date: Tue, 15 Jul 2025 16:42:31 +0200 Subject: [PATCH 1/2] utils: Improve rule priority support in is_covered/is_equal MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - `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 --- utils/apparmor/rule/__init__.py | 10 +++++++--- 1 file changed, 7 insertions(+), 3 deletions(-) diff --git a/utils/apparmor/rule/__init__.py b/utils/apparmor/rule/__init__.py index 8484efecd..1f5030fe8 100644 --- a/utils/apparmor/rule/__init__.py +++ b/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 ): From f78aa365475ccdc91e8176834c903761a9a6c592 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Maxime=20B=C3=A9lair?= Date: Tue, 15 Jul 2025 17:07:55 +0200 Subject: [PATCH 2/2] Add tests for priority is_covered/is_equal fix MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Maxime Bélair --- utils/test/test-dbus.py | 11 +++++++++++ utils/test/test-io_uring.py | 16 ++++++++++++++++ utils/test/test-unix.py | 7 +++++++ 3 files changed, 34 insertions(+) diff --git a/utils/test/test-dbus.py b/utils/test/test-dbus.py index d31dff00e..afe7c73c5 100644 --- a/utils/test/test-dbus.py +++ b/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 diff --git a/utils/test/test-io_uring.py b/utils/test/test-io_uring.py index 7e09b1ce9..56f4a8f65 100644 --- a/utils/test/test-io_uring.py +++ b/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 = ( diff --git a/utils/test/test-unix.py b/utils/test/test-unix.py index 7d6024765..1dcd11ae5 100644 --- a/utils/test/test-unix.py +++ b/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 = (