Merge utils: fix coding style in mount

Signed-off-by: Georgia Garcia <georgia.garcia@canonical.com>

MR: https://gitlab.com/apparmor/apparmor/-/merge_requests/1173
Approved-by: Christian Boltz <apparmor@cboltz.de>
Merged-by: Christian Boltz <apparmor@cboltz.de>
This commit is contained in:
Christian Boltz 2024-03-04 13:05:41 +00:00
commit 6695944c2c
2 changed files with 116 additions and 118 deletions

View file

@ -48,28 +48,29 @@ flags_keywords = [
join_valid_flags = '|'.join(flags_keywords)
join_valid_fs = '|'.join(valid_fs)
sep = r"\s*[\s,]\s*"
sep = r'\s*[\s,]\s*'
fs_type_pattern = r"\b(?P<fstype_or_vfstype>fstype|vfstype)\b\s*(?P<fstype_equals_or_in>=|in)\s*"\
r"(?P<fstype>\(\s*(" + join_valid_fs + ")(" + sep + "(" + join_valid_fs + r"))*\s*\)|"\
r"\{\s*(" + join_valid_fs + ")(" + sep + "(" + join_valid_fs + r"))*\s*\}|(\s*" + join_valid_fs + "))"\
fs_type_pattern = r'\b(?P<fstype_or_vfstype>fstype|vfstype)\b\s*(?P<fstype_equals_or_in>=|in)\s*'\
r'(?P<fstype>\(\s*(' + join_valid_fs + r')(' + sep + r'(' + join_valid_fs + r'))*\s*\)|'\
r'\{\s*(' + join_valid_fs + r')(' + sep + r'(' + join_valid_fs + r'))*\s*\}|(\s*' + join_valid_fs + r'))'\
option_pattern = r"\s*(\boption(s?)\b\s*(?P<options_equals_or_in>=|in)\s*"\
r"(?P<options>\(\s*(" + join_valid_flags + ")(" + sep + "(" + join_valid_flags + r"))*\s*\)|" \
r"(\s*" + join_valid_flags + ")"\
"))?"
mount_condition_pattern = rf"({fs_type_pattern})?\s*({option_pattern})?"
option_pattern = r'\s*(\boption(s?)\b\s*(?P<options_equals_or_in>=|in)\s*'\
r'(?P<options>\(\s*(' + join_valid_flags + r')(' + sep + r'(' + join_valid_flags + r'))*\s*\)|' \
r'(\s*' + join_valid_flags + r')'\
r'))?'
mount_condition_pattern = rf'({fs_type_pattern})?\s*({option_pattern})?'
# Source can either be
# - A path : /foo
# - A filesystem : sysfs (sudo mount -t tmpfs tmpfs /tmp/bar)
# - Any label : mntlabel (sudo mount -t tmpfs mntlabel /tmp/bar)
source_fileglob_pattern = r"(\s*" + (RE_PROFILE_PATH_OR_VAR % "source_file")[:-1] + "|" + r"\w+" + "))"
dest_fileglob_pattern = r"(\s*" + RE_PROFILE_PATH_OR_VAR % "dest_file" + ')'
source_fileglob_pattern = r'(\s*' + (RE_PROFILE_PATH_OR_VAR % 'source_file')[:-1] + r'|' + r'\w+' + r'))'
dest_fileglob_pattern = r'(\s*' + RE_PROFILE_PATH_OR_VAR % 'dest_file' + r')'
RE_MOUNT_DETAILS = re.compile(r'^\s*' + mount_condition_pattern + rf'(\s+{source_fileglob_pattern})?' + rf'(\s+->\s+{dest_fileglob_pattern})?\s*' + r'$')
RE_UMOUNT_DETAILS = re.compile(r'^\s*' + mount_condition_pattern + rf'(\s+{dest_fileglob_pattern})?\s*' + r'$')
RE_MOUNT_DETAILS = re.compile(r"^\s*" + mount_condition_pattern + rf"(\s+{source_fileglob_pattern})?" + rf"(\s+->\s+{dest_fileglob_pattern})?\s*" +"$")
RE_UMOUNT_DETAILS = re.compile(r"^\s*" + mount_condition_pattern + rf"(\s+{dest_fileglob_pattern})?\s*" +"$")
class MountRule(BaseRule):
'''Class to handle and store a single mount rule'''
@ -105,9 +106,9 @@ class MountRule(BaseRule):
self.source, self.all_source = self._aare_or_all(source, 'source', is_path=False, log_event=log_event)
if not self.all_fstype and self.is_fstype_equal not in ("=", "in"):
if not self.all_fstype and self.is_fstype_equal not in ('=', 'in'):
raise AppArmorBug(f'Invalid is_fstype_equal : {self.is_fstype_equal}')
if not self.all_options and self.is_options_equal not in ("=", "in"):
if not self.all_options and self.is_options_equal not in ('=', 'in'):
raise AppArmorBug(f'Invalid is_options_equal : {self.is_options_equal}')
if self.operation != 'mount' and not self.all_source:
raise AppArmorException(f'Operation {self.operation} cannot have a source')
@ -120,7 +121,6 @@ class MountRule(BaseRule):
self.can_glob = not self.all_source and not self.all_dest and not self.all_options
@classmethod
def _create_instance(cls, raw_rule, matches):
'''parse raw_rule and return instance of this class'''
@ -133,7 +133,7 @@ class MountRule(BaseRule):
if matches.group('details'):
rule_details = matches.group('details')
if operation == "mount":
if operation == 'mount':
parsed = RE_MOUNT_DETAILS.search(rule_details)
else:
parsed = RE_UMOUNT_DETAILS.search(rule_details)
@ -182,19 +182,19 @@ class MountRule(BaseRule):
fstype = ' fstype%s(%s)' % (wrap_in_with_spaces(self.is_fstype_equal), ', '.join(sorted(self.fstype))) if not self.all_fstype else ''
options = ' options%s(%s)' % (wrap_in_with_spaces(self.is_options_equal), ', '.join(sorted(self.options))) if not self.all_options else ''
source = ""
dest = ""
source = ''
dest = ''
if self.operation == 'mount':
if not self.all_source:
source = " " + str(self.source.regex)
source = ' ' + str(self.source.regex)
if not self.all_dest:
dest = " -> " + str(self.dest.regex)
dest = ' -> ' + str(self.dest.regex)
else:
if not self.all_dest:
dest = " " + str(self.dest.regex)
dest = ' ' + str(self.dest.regex)
return ('%s%s%s%s%s%s%s,%s' % (self.modifiers_str(),
space,
@ -275,6 +275,7 @@ class MountRule(BaseRule):
)
class MountRuleset(BaseRuleset):
'''Class to handle and store a collection of Mount rules'''

View file

@ -29,35 +29,33 @@ class MountTestParse(AATest):
tests = (
# Rule Operation Filesystem Options Source Destination Audit Deny Allow Comment
('mount fstype=bpf options=rw bpf -> /sys/fs/bpf/,', MountRule('mount', ('=',('bpf')), ('=',('rw')), "bpf", "/sys/fs/bpf/", False, False, False, '' )),
('mount fstype=bpf options=(rw) random_label -> /sys/fs/bpf/,', MountRule('mount', ('=',('bpf')), ('=',('rw')), "random_label", "/sys/fs/bpf/", False, False, False, '' )),
('mount fstype=bpf options=rw bpf -> /sys/fs/bpf/,', MountRule('mount', ('=', ('bpf')), ('=', ('rw')), 'bpf', '/sys/fs/bpf/', False, False, False, '' )),
('mount fstype=bpf options=(rw) random_label -> /sys/fs/bpf/,', MountRule('mount', ('=', ('bpf')), ('=', ('rw')), 'random_label', '/sys/fs/bpf/', False, False, False, '' )),
('mount,', MountRule('mount', MountRule.ALL, MountRule.ALL, MountRule.ALL, MountRule.ALL, False, False, False, '' )),
('mount fstype=(ext3, ext4),', MountRule('mount', ('=', ('ext3', 'ext4')), MountRule.ALL, MountRule.ALL, MountRule.ALL, False, False, False, '' )),
('mount bpf,', MountRule('mount', MountRule.ALL, MountRule.ALL, "bpf", MountRule.ALL, False, False, False, '' )),
('mount none,', MountRule('mount', MountRule.ALL, MountRule.ALL, "none", MountRule.ALL, False, False, False, '' )),
('mount bpf,', MountRule('mount', MountRule.ALL, MountRule.ALL, 'bpf', MountRule.ALL, False, False, False, '' )),
('mount none,', MountRule('mount', MountRule.ALL, MountRule.ALL, 'none', MountRule.ALL, False, False, False, '' )),
('mount fstype=(ext3, ext4) options=(ro),', MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), MountRule.ALL, MountRule.ALL, False, False, False, '' )),
('mount @{mntpnt},', MountRule('mount', MountRule.ALL, MountRule.ALL, "@{mntpnt}", MountRule.ALL, False, False, False, '' )),
('mount /a,', MountRule('mount', MountRule.ALL, MountRule.ALL, "/a", MountRule.ALL, False, False, False, '' )),
('mount fstype=(ext3, ext4) /a -> /b,', MountRule('mount', ('=',('ext3', 'ext4')), MountRule.ALL, "/a", "/b", False, False, False, '' )),
('mount fstype=(ext3, ext4) options=(ro, rbind) /a -> /b,', MountRule('mount', ('=',('ext3', 'ext4')), ('=',('ro', 'rbind')), "/a", "/b", False, False, False, '' )),
('mount fstype=(ext3, ext4) options=(ro, rbind) /a -> /b, #cmt', MountRule('mount', ('=',('ext3', 'ext4')), ('=',('ro', 'rbind')), "/a", "/b", False, False, False, ' #cmt')),
('mount fstype=(ext3, ext4) options in (ro, rbind) /a -> /b,', MountRule('mount', ('=',('ext3', 'ext4')), ('in',('ro', 'rbind')), "/a", "/b", False, False, False, '' )),
('mount fstype in (ext3, ext4) options=(ro, rbind) /a -> /b, #cmt', MountRule('mount', ('in',('ext3', 'ext4')),('=',('ro', 'rbind')), "/a", "/b", False, False, False, ' #cmt')),
('mount fstype in (ext3, ext4) option in (ro, rbind) /a, #cmt', MountRule('mount', ('in',('ext3', 'ext4')),('in',('ro', 'rbind')), "/a", MountRule.ALL, False, False, False, ' #cmt')),
('mount fstype=(ext3, ext4) option=(ro, rbind) /a -> /b, #cmt', MountRule('mount', ('=',('ext3', 'ext4')), ('=', ('ro', 'rbind')), "/a", "/b", False, False, False, ' #cmt')),
('mount @{mntpnt},', MountRule('mount', MountRule.ALL, MountRule.ALL, '@{mntpnt}', MountRule.ALL, False, False, False, '' )),
('mount /a,', MountRule('mount', MountRule.ALL, MountRule.ALL, '/a', MountRule.ALL, False, False, False, '' )),
('mount fstype=(ext3, ext4) /a -> /b,', MountRule('mount', ('=', ('ext3', 'ext4')), MountRule.ALL, '/a', '/b', False, False, False, '' )),
('mount fstype=(ext3, ext4) options=(ro, rbind) /a -> /b,', MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro', 'rbind')), '/a', '/b', False, False, False, '' )),
('mount fstype=(ext3, ext4) options=(ro, rbind) /a -> /b, #cmt', MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro', 'rbind')), '/a', '/b', False, False, False, ' #cmt')),
('mount fstype=(ext3, ext4) options in (ro, rbind) /a -> /b,', MountRule('mount', ('=', ('ext3', 'ext4')), ('in', ('ro', 'rbind')), '/a', '/b', False, False, False, '' )),
('mount fstype in (ext3, ext4) options=(ro, rbind) /a -> /b, #cmt', MountRule('mount', ('in', ('ext3', 'ext4')), ('=', ('ro', 'rbind')), '/a', '/b', False, False, False, ' #cmt')),
('mount fstype in (ext3, ext4) option in (ro, rbind) /a, #cmt', MountRule('mount', ('in', ('ext3', 'ext4')), ('in', ('ro', 'rbind')), '/a', MountRule.ALL, False, False, False, ' #cmt')),
('mount fstype=(ext3, ext4) option=(ro, rbind) /a -> /b, #cmt', MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro', 'rbind')), '/a', '/b', False, False, False, ' #cmt')),
('mount options=(rw, rbind) /usr/lib{,32,64,x32}/modules/ -> /tmp/snap.rootfs_*{,/usr}/lib/modules/,',
MountRule('mount', MountRule.ALL, ('=',('rw', 'rbind')), "/usr/lib{,32,64,x32}/modules/",
"/tmp/snap.rootfs_*{,/usr}/lib/modules/",
MountRule('mount', MountRule.ALL, ('=', ('rw', 'rbind')), '/usr/lib{,32,64,x32}/modules/',
'/tmp/snap.rootfs_*{,/usr}/lib/modules/',
False, False, False, '' )),
('umount,', MountRule('umount', MountRule.ALL, MountRule.ALL, MountRule.ALL, MountRule.ALL, False, False, False, '' )),
('umount fstype=ext3,', MountRule('umount', ('=', ('ext3')), MountRule.ALL, MountRule.ALL, MountRule.ALL, False, False, False, '' )),
('umount /a,', MountRule('umount', MountRule.ALL, MountRule.ALL, MountRule.ALL, "/a", False, False, False, '' )),
('umount /a,', MountRule('umount', MountRule.ALL, MountRule.ALL, MountRule.ALL, '/a', False, False, False, '' )),
('remount,', MountRule('remount', MountRule.ALL, MountRule.ALL, MountRule.ALL, MountRule.ALL, False, False, False, '' )),
('remount fstype=ext4,', MountRule('remount', ('=', ('ext4')), MountRule.ALL, MountRule.ALL, MountRule.ALL, False, False, False, '' )),
('remount /b,', MountRule('remount',MountRule.ALL, MountRule.ALL, MountRule.ALL, "/b", False, False, False, '' )),
('remount /b,', MountRule('remount', MountRule.ALL, MountRule.ALL, MountRule.ALL, '/b', False, False, False, '' )),
)
def _run_test(self, rawrule, expected):
@ -88,7 +86,7 @@ class MountTestParseInvalid(AATest):
def test_diff_non_mountrule(self):
exp = namedtuple('exp', ('audit', 'deny'))
obj = MountRule('mount',("=", 'ext4'), MountRule.ALL, MountRule.ALL, MountRule.ALL)
obj = MountRule('mount', ('=', 'ext4'), MountRule.ALL, MountRule.ALL, MountRule.ALL)
with self.assertRaises(AppArmorBug):
obj.is_equal(exp(False, False), False)
@ -109,42 +107,42 @@ class MountTestParseInvalid(AATest):
MountRule('mount', MountRule.ALL, ('=', 'invalid'), MountRule.ALL, MountRule.ALL) # fstype[0] should be '=' or 'in'
def test_diff_fstype(self):
obj1 = MountRule('mount',("=", 'ext4'), MountRule.ALL, MountRule.ALL, MountRule.ALL)
obj1 = MountRule('mount', ('=', 'ext4'), MountRule.ALL, MountRule.ALL, MountRule.ALL)
obj2 = MountRule('mount', MountRule.ALL, MountRule.ALL, MountRule.ALL, MountRule.ALL)
self.assertFalse(obj1.is_equal(obj2, False))
def test_diff_source(self):
obj1 = MountRule('mount',MountRule.ALL, MountRule.ALL, "/foo", MountRule.ALL)
obj2 = MountRule('mount',MountRule.ALL, MountRule.ALL, "/bar", MountRule.ALL)
obj1 = MountRule('mount', MountRule.ALL, MountRule.ALL, '/foo', MountRule.ALL)
obj2 = MountRule('mount', MountRule.ALL, MountRule.ALL, '/bar', MountRule.ALL)
self.assertFalse(obj1.is_equal(obj2, False))
def test_invalid_umount_with_source(self):
with self.assertRaises(AppArmorException):
MountRule('umount', MountRule.ALL, MountRule.ALL, "/foo", MountRule.ALL) # Umount and remount shall not have a source
MountRule('umount', MountRule.ALL, MountRule.ALL, '/foo', MountRule.ALL) # Umount and remount shall not have a source
def test_invalid_remount_with_source(self):
with self.assertRaises(AppArmorException):
MountRule('remount', MountRule.ALL, MountRule.ALL, "/foo", MountRule.ALL)
MountRule('remount', MountRule.ALL, MountRule.ALL, '/foo', MountRule.ALL)
class MountTestFilesystems(AATest):
def test_fs(self):
with open("/proc/filesystems") as f:
with open('/proc/filesystems') as f:
for line in f:
fs_name = line.split()[-1]
self.assertTrue(fs_name in valid_fs, "/proc/filesystems contains %s which is not listed in MountRule valid_fs" % fs_name)
self.assertTrue(fs_name in valid_fs, '/proc/filesystems contains %s which is not listed in MountRule valid_fs' % fs_name)
class MountTestGlob(AATest):
def test_glob(self):
globList = [(
"mount options=(bind, rw) /home/user/Downloads/ -> /mnt/a/,",
"mount options=(bind, rw) /home/user/Downloads/,",
"mount options=(bind, rw) /home/user/*/,",
"mount options=(bind, rw) /home/**/,",
"mount options=(bind, rw),",
"mount,",
"mount,",
'mount options=(bind, rw) /home/user/Downloads/ -> /mnt/a/,',
'mount options=(bind, rw) /home/user/Downloads/,',
'mount options=(bind, rw) /home/user/*/,',
'mount options=(bind, rw) /home/**/,',
'mount options=(bind, rw),',
'mount,',
'mount,',
)]
for globs in globList:
for i in range(len(globs) - 1):
@ -180,11 +178,12 @@ class MountTestClean(AATest):
self.assertEqual(expected, clean, 'unexpected clean rule')
self.assertEqual(rawrule.strip(), raw, 'unexpected raw rule')
class MountLogprofHeaderTest(AATest):
tests = (
('mount,', [_('Operation'), _('mount'), _('Fstype'), _('ALL'), _('Options'), _('ALL'), _('Source'), _('ALL'), _('Destination'), _('ALL')]),
('mount options=(ro, nosuid) /a,', [_('Operation'), _('mount'), _('Fstype'), _('ALL'), _('Options'), ("=", _('nosuid ro')),_('Source'), _('/a'), _('Destination'), _('ALL')]),
('mount fstype=(ext3, ext4) options=(ro, nosuid) /a -> /b,', [_('Operation'), _('mount'), _('Fstype'), ("=", _('ext3 ext4')),_('Options'), ("=", _('nosuid ro')),_('Source'), _('/a'), _('Destination'), _('/b')])
('mount options=(ro, nosuid) /a,', [_('Operation'), _('mount'), _('Fstype'), _('ALL'), _('Options'), ('=', _('nosuid ro')), _('Source'), _('/a'), _('Destination'), _('ALL')]),
('mount fstype=(ext3, ext4) options=(ro, nosuid) /a -> /b,', [_('Operation'), _('mount'), _('Fstype'), ('=', _('ext3 ext4')), _('Options'), ('=', _('nosuid ro')), _('Source'), _('/a'), _('Destination'), _('/b')])
)
def _run_test(self, params, expected):
@ -194,47 +193,45 @@ class MountLogprofHeaderTest(AATest):
class MountIsCoveredTest(AATest):
def test_is_covered(self):
obj = MountRule("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "/foo/b*", "/b*")
obj = MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), '/foo/b*', '/b*')
tests = [
("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "/foo/b", "/bar"),
("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "/foo/bar", "/b")
('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), '/foo/b', '/bar'),
('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), '/foo/bar', '/b')
]
for test in tests:
self.assertTrue(obj.is_covered(MountRule(*test)))
self.assertFalse(obj.is_equal(MountRule(*test)))
def test_is_covered_fs_source(self):
obj = MountRule("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "tmpfs", MountRule.ALL)
self.assertTrue(obj.is_covered(MountRule("mount", ("=", ('ext3')), ("=", ('ro')), "tmpfs", MountRule.ALL)))
self.assertFalse(obj.is_equal(MountRule("mount", ("=", ('ext3')), ("=", ('ro')), "tmpfs", MountRule.ALL)))
obj = MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), 'tmpfs', MountRule.ALL)
self.assertTrue(obj.is_covered(MountRule('mount', ('=', ('ext3')), ('=', ('ro')), 'tmpfs', MountRule.ALL)))
self.assertFalse(obj.is_equal(MountRule('mount', ('=', ('ext3')), ('=', ('ro')), 'tmpfs', MountRule.ALL)))
def test_is_notcovered(self):
obj = MountRule("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "/foo/b*", "/b*")
obj = MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), '/foo/b*', '/b*')
tests = [
("mount", ("in", ('ext3', 'ext4')), ("=", ('ro')), "/foo/bar", "/bar" ),
("mount", ("=", ('procfs', 'ext4')), ("=", ('ro')), "/foo/bar", "/bar" ),
("mount", ("=", ('ext3')), ("=", ('rw')), "/foo/bar", "/bar" ),
("mount", ("=", ('ext3', 'ext4')), MountRule.ALL, "/foo/b*", "/bar" ),
("mount", MountRule.ALL, ("=", ('ro')), "/foo/b*", "/bar" ),
("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "/invalid/bar", "/bar" ),
("umount", MountRule.ALL, MountRule.ALL, MountRule.ALL, "/bar" ),
("remount", MountRule.ALL, MountRule.ALL, MountRule.ALL, "/bar" ),
("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "tmpfs", "/bar" ),
("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "/foo/b*", "/invalid" ),
('mount', ('in', ('ext3', 'ext4')), ('=', ('ro')), '/foo/bar', '/bar' ),
('mount', ('=', ('procfs', 'ext4')), ('=', ('ro')), '/foo/bar', '/bar' ),
('mount', ('=', ('ext3')), ('=', ('rw')), '/foo/bar', '/bar' ),
('mount', ('=', ('ext3', 'ext4')), MountRule.ALL, '/foo/b*', '/bar' ),
('mount', MountRule.ALL, ('=', ('ro')), '/foo/b*', '/bar' ),
('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), '/invalid/bar', '/bar' ),
('umount', MountRule.ALL, MountRule.ALL, MountRule.ALL, '/bar' ),
('remount', MountRule.ALL, MountRule.ALL, MountRule.ALL, '/bar' ),
('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), 'tmpfs', '/bar' ),
('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), '/foo/b*', '/invalid'),
]
for test in tests:
self.assertFalse(obj.is_covered(MountRule(*test)))
self.assertFalse(obj.is_equal(MountRule(*test)))
def test_is_not_covered_fs_source(self):
obj = MountRule("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "tmpfs", MountRule.ALL)
test = ("mount", ("=", ('ext3', 'ext4')), ("=", ('ro')), "procfs", MountRule.ALL)
obj = MountRule('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), 'tmpfs', MountRule.ALL)
test = ('mount', ('=', ('ext3', 'ext4')), ('=', ('ro')), 'procfs', MountRule.ALL)
self.assertFalse(obj.is_covered(MountRule(*test)))
self.assertFalse(obj.is_equal(MountRule(*test)))
setup_all_loops(__name__)
if __name__ == '__main__':
unittest.main(verbosity=1)