2
0
mirror of https://gitlab.com/apparmor/apparmor synced 2025-08-22 18:17:09 +00:00

utils: add support for multiple fstypes in mount rules

Signed-off-by: Georgia Garcia <georgia.garcia@canonical.com>
This commit is contained in:
Georgia Garcia 2025-05-22 15:51:45 -03:00
parent 79f2ea72b0
commit 2e7da63183
3 changed files with 98 additions and 70 deletions

View File

@ -124,18 +124,26 @@ class MountRule(BaseRule):
self.operation = operation self.operation = operation
if fstype == self.ALL or fstype[1] == self.ALL: if not isinstance(fstype, list):
self.all_fstype = True fstype = [fstype]
self.fstype = None
self.is_fstype_equal = None # self.all_fstype will only be true if no fstypes are
else: # specified, so it's fine to set it inside the loop
self.all_fstype = False self.fstype = []
for it in fstype[1]: for fst in fstype:
aare_len, unused = parse_aare(it, 0, 'fstype') if fst == self.ALL or fst[1] == self.ALL:
if aare_len != len(it): self.all_fstype = True
raise AppArmorException(f'Invalid aare : {it}') fstype_values = None
self.fstype = fstype[1] is_fstype_equal = None
self.is_fstype_equal = fstype[0] else:
self.all_fstype = False
for it in fst[1]:
aare_len, unused = parse_aare(it, 0, 'fstype')
if aare_len != len(it):
raise AppArmorException(f'Invalid aare : {it}')
fstype_values = fst[1]
is_fstype_equal = fst[0]
self.fstype.append(MountConditional('fstype', fstype_values, self.all_fstype, is_fstype_equal, 'aare'))
if not isinstance(options, list): if not isinstance(options, list):
options = [options] options = [options]
@ -148,13 +156,11 @@ class MountRule(BaseRule):
if unknown_items: if unknown_items:
raise AppArmorException(_('Passed unknown options keyword to %s: %s') % (type(self).__name__, ' '.join(unknown_items))) raise AppArmorException(_('Passed unknown options keyword to %s: %s') % (type(self).__name__, ' '.join(unknown_items)))
is_options_equal = opts[0] if not self.all_options else None is_options_equal = opts[0] if not self.all_options else None
self.options.append(MountConditional('options', opt_values, self.all_options, is_options_equal)) self.options.append(MountConditional('options', opt_values, self.all_options, is_options_equal, 'list'))
self.source, self.all_source = self._aare_or_all(source, 'source', is_path=False, log_event=log_event, empty_ok=True) self.source, self.all_source = self._aare_or_all(source, 'source', is_path=False, log_event=log_event, empty_ok=True)
self.dest, self.all_dest = self._aare_or_all(dest, 'dest', is_path=False, log_event=log_event) self.dest, self.all_dest = self._aare_or_all(dest, 'dest', is_path=False, log_event=log_event)
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 self.operation != 'mount' and not self.all_source: if self.operation != 'mount' and not self.all_source:
raise AppArmorException(f'Operation {self.operation} cannot have a source') raise AppArmorException(f'Operation {self.operation} cannot have a source')
@ -191,17 +197,12 @@ class MountRule(BaseRule):
if not r: if not r:
raise AppArmorException('Can\'t parse mount rule ' + raw_rule) raise AppArmorException('Can\'t parse mount rule ' + raw_rule)
fstype = (None, cls.ALL)
if r['fstype'] is not None: if r['fstype'] is not None:
# mount rules with multiple 'fstype' are not supported by the tools yet, and when writing them, only the last 'fstype' would survive. fstype = []
# Therefore raise an exception when parsing such a rule to prevent breaking the rule. for m in re.finditer(fs_type_pattern, rule_details):
if RE_MOUNT_MULTIPLE_FS_TYPE.search(raw_rule): fst = parse_aare_list(strip_parenthesis(m.group('fstype')), 'fstype')
raise AppArmorException("mount rules with multiple 'fstype' are not supported by the tools") fstype.append((m.group('fstype_equals_or_in'), fst))
is_fstype_equal = r['fstype_equals_or_in']
fstype = parse_aare_list(strip_parenthesis(r['fstype']), 'fstype')
else:
is_fstype_equal = None
fstype = cls.ALL
opts = (None, cls.ALL) opts = (None, cls.ALL)
if r['options'] is not None: if r['options'] is not None:
@ -222,19 +223,21 @@ class MountRule(BaseRule):
else: else:
opts = (None, cls.ALL) opts = (None, cls.ALL)
is_fstype_equal = None fstype = (None, cls.ALL)
fstype = cls.ALL
source = cls.ALL source = cls.ALL
dest = cls.ALL dest = cls.ALL
return cls(operation=operation, fstype=(is_fstype_equal, fstype), options=opts, return cls(operation=operation, fstype=fstype, options=opts,
source=source, dest=dest, audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, source=source, dest=dest, audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment,
priority=priority) priority=priority)
def get_clean(self, depth=0): def get_clean(self, depth=0):
space = ' ' * depth space = ' ' * depth
fstype = ' fstype%s(%s)' % (wrap_in_with_spaces(self.is_fstype_equal), ', '.join(sorted(self.fstype))) if not self.all_fstype else '' fstype = ''
for fst in self.fstype:
fstype += fst.get_clean()
options = '' options = ''
for opt in self.options: for opt in self.options:
options += opt.get_clean() options += opt.get_clean()
@ -277,17 +280,8 @@ class MountRule(BaseRule):
def _is_covered_localvars(self, other_rule): def _is_covered_localvars(self, other_rule):
if self.operation != other_rule.operation: if self.operation != other_rule.operation:
return False return False
if self.is_fstype_equal != other_rule.is_fstype_equal: if not self._is_cond_list_covered(self.fstype, other_rule.fstype):
return False return False
for o_it in other_rule.fstype or []:
found = False
for s_it in self.fstype or []:
if self._is_covered_aare(AARE(s_it, False), self.all_fstype, AARE(o_it, False), other_rule.all_fstype, 'fstype'):
found = True
if not found:
return False
if not self._is_cond_list_covered(self.options, other_rule.options): if not self._is_cond_list_covered(self.options, other_rule.options):
return False return False
if not self._is_covered_aare(self.source, self.all_source, other_rule.source, other_rule.all_source, 'source'): if not self._is_covered_aare(self.source, self.all_source, other_rule.source, other_rule.all_source, 'source'):
@ -300,8 +294,6 @@ class MountRule(BaseRule):
def _is_equal_localvars(self, rule_obj, strict): def _is_equal_localvars(self, rule_obj, strict):
if self.operation != rule_obj.operation: if self.operation != rule_obj.operation:
return False return False
if self.is_fstype_equal != rule_obj.is_fstype_equal:
return False
if self.fstype != rule_obj.fstype or self.options != rule_obj.options: if self.fstype != rule_obj.fstype or self.options != rule_obj.options:
return False return False
if not self._is_equal_aare(self.source, self.all_source, rule_obj.source, rule_obj.all_source, 'source'): if not self._is_equal_aare(self.source, self.all_source, rule_obj.source, rule_obj.all_source, 'source'):
@ -352,7 +344,12 @@ class MountRule(BaseRule):
def _logprof_header_localvars(self): def _logprof_header_localvars(self):
operation = self.operation operation = self.operation
fstype = logprof_value_or_all(self.fstype, self.all_fstype)
fstype_output = ()
for fst in self.fstype:
fstype = logprof_value_or_all(fst.values, fst.all_values)
fstype_output = fstype_output + (_('Fstype'), (fst.operator, fstype) if fstype != 'ALL' else fstype)
opts_output = () opts_output = ()
for opt in self.options: for opt in self.options:
options = logprof_value_or_all(opt.values, opt.all_values) options = logprof_value_or_all(opt.values, opt.all_values)
@ -362,7 +359,7 @@ class MountRule(BaseRule):
return ( return (
_('Operation'), operation, _('Operation'), operation,
_('Fstype'), (self.is_fstype_equal, fstype) if fstype != 'ALL' else fstype, *fstype_output,
*opts_output, *opts_output,
_('Source'), source, _('Source'), source,
_('Destination'), dest, _('Destination'), dest,
@ -416,11 +413,13 @@ def wrap_in_with_spaces(value):
class MountConditional(MountRule): class MountConditional(MountRule):
'''Class to handle and store mount conditionals''' '''Class to handle and store mount conditionals'''
def __init__(self, name, values, all_values, operator): def __init__(self, name, values, all_values, operator, cond_type=None):
self.name = name self.name = name
self.values = values self.values = values
self.all_values = all_values self.all_values = all_values
self.operator = operator self.operator = operator
self.cond_type = cond_type
self.raw_rule = '' # needed so __repr__ calls get_clean
if not self.all_values and self.operator not in ('=', 'in'): if not self.all_values and self.operator not in ('=', 'in'):
raise AppArmorBug(f'Invalid operator for {self.name}: {self.operator}') raise AppArmorBug(f'Invalid operator for {self.name}: {self.operator}')
@ -432,6 +431,9 @@ class MountConditional(MountRule):
if self.all_values != other.all_values: if self.all_values != other.all_values:
return False return False
if self.cond_type != other.cond_type:
return False
if self.values != other.values: if self.values != other.values:
return False return False
@ -450,13 +452,27 @@ class MountConditional(MountRule):
if self.all_values: if self.all_values:
return True return True
if self._is_covered_list(self.values, self.all_values, other.values, other.all_values, self.name): if other.all_values:
if self.operator == other.operator: return False
return True
if self.cond_type == 'list':
if not self._is_covered_list(self.values, self.all_values, other.values, other.all_values, self.name):
return False
elif self.cond_type == 'aare': # list of aares - all values in other must be at least once in self.values
if not all(any(self._is_covered_aare(AARE(value, False), self.all_values,
AARE(other_value, False), other.all_values, self.name)
for value in self.values)
for other_value in other.values):
return False
else:
raise AppArmorBug('Type should only be empty if ALL is true')
if self.operator == other.operator:
return True
return False return False
def get_clean(self) -> str: def get_clean(self, depth=0) -> str:
conditional = '' conditional = ''
if not self.all_values: if not self.all_values:
conditional += ' %s%s(%s)' % (self.name, wrap_in_with_spaces(self.operator), ', '.join(sorted(self.values))) conditional += ' %s%s(%s)' % (self.name, wrap_in_with_spaces(self.operator), ', '.join(sorted(self.values)))

View File

@ -62,19 +62,32 @@ class MountTestParse(AATest):
('mount options=(runbindable, rw) -> /,', MountRule('mount', MountRule.ALL, ('=', ['runbindable', 'rw']), MountRule.ALL, '/', False, False, False, '')), ('mount options=(runbindable, rw) -> /,', MountRule('mount', MountRule.ALL, ('=', ['runbindable', 'rw']), MountRule.ALL, '/', False, False, False, '')),
('mount "" -> /,', MountRule('mount', MountRule.ALL, MountRule.ALL, '', '/', False, False, False, '')), ('mount "" -> /,', MountRule('mount', MountRule.ALL, MountRule.ALL, '', '/', False, False, False, '')),
('mount options=(ro) options=(rw) fstype=ext4 -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('=', ('rw'))], # noqa: E127 ('mount options=(ro) options=(rw) fstype=ext4 -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('=', ('rw'))], # noqa: E127
MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127 MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount options=(ro) fstype=ext4 options=(rw) /src -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('=', ('rw'))], # noqa: E127 ('mount options=(ro) fstype=ext4 options=(rw) /src -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('=', ('rw'))], # noqa: E127
'/src', '/dest', False, False, False, '')), # noqa: E127 '/src', '/dest', False, False, False, '')), # noqa: E127
('mount options in (ro) options in (rw) fstype=ext4 -> /dest,', MountRule('mount', ('=', ['ext4']), [('in', ('ro')), ('in', ('rw'))], # noqa: E127 ('mount options in (ro) options in (rw) fstype=ext4 -> /dest,', MountRule('mount', ('=', ['ext4']), [('in', ('ro')), ('in', ('rw'))], # noqa: E127
MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127 MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount options in (ro) fstype=ext4 options in (rw) -> /dest,', MountRule('mount', ('=', ['ext4']), [('in', ('ro')), ('in', ('rw'))], # noqa: E127 ('mount options in (ro) fstype=ext4 options in (rw) -> /dest,', MountRule('mount', ('=', ['ext4']), [('in', ('ro')), ('in', ('rw'))], # noqa: E127
MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127 MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount options = (ro) options in (rw) fstype=ext4 -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('in', ('rw'))], # noqa: E127 ('mount options = (ro) options in (rw) fstype=ext4 -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('in', ('rw'))], # noqa: E127
MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127 MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount options = (ro) fstype=ext4 options in (rw) -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('in', ('rw'))], # noqa: E127 ('mount options = (ro) fstype=ext4 options in (rw) -> /dest,', MountRule('mount', ('=', ['ext4']), [('=', ('ro')), ('in', ('rw'))], # noqa: E127
MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127 MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount options=(ro) fstype=ext3 fstype=ext4 -> /dest,', MountRule('mount', [('=', ['ext3']), ('=', ['ext4'])],
('=', ('ro')), MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount fstype=ext3 options=(ro) fstype=ext4 src -> /dest,', MountRule('mount', [('=', ['ext3']), ('=', ['ext4'])],
('=', ('ro')), 'src', '/dest', False, False, False, '')), # noqa: E127
('mount options=(ro) fstype in (ext3) fstype in (ext4) -> /dest,', MountRule('mount', [('in', ['ext3']), ('in', ['ext4'])],
('=', ('ro')), MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount fstype in (ext3) options=(ro) fstype in (ext4) -> /dest,', MountRule('mount', [('in', ['ext3']), ('in', ['ext4'])],
('=', ('ro')), MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount options=(ro) fstype in (ext3) fstype=(ext4) -> /dest,', MountRule('mount', [('in', ['ext3']), ('=', ['ext4'])],
('=', ('ro')), MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('mount fstype = (ext3) options=(ro) fstype in ext4 -> /dest,', MountRule('mount', [('=', ['ext3']), ('in', ['ext4'])],
('=', ('ro')), MountRule.ALL, '/dest', False, False, False, '')), # noqa: E127
('umount,', MountRule('umount', MountRule.ALL, MountRule.ALL, MountRule.ALL, MountRule.ALL, 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 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, '')),
@ -109,14 +122,6 @@ class MountTestParseInvalid(AATest):
('mount options=(),', AppArmorException), ('mount options=(),', AppArmorException),
('mount option=(invalid),', AppArmorException), ('mount option=(invalid),', AppArmorException),
('mount option=(ext3ext4),', AppArmorException), ('mount option=(ext3ext4),', AppArmorException),
# mount rules with multiple 'fstype' are not supported by the tools yet, and when writing them, only the last 'fstype' would survive. Therefore MountRule intentionally raises an exception when parsing such a rule.
('mount options=(ro) fstype=ext3 fstype=ext4 -> /destination,', AppArmorException),
('mount fstype=ext3 options=(ro) fstype=ext4 -> /destination,', AppArmorException),
('mount options=(ro) fstype in (ext3) fstype in (ext4) -> /destination,', AppArmorException),
('mount fstype in (ext3) options=(ro) fstype in (ext4) -> /destination,', AppArmorException),
('mount options=(ro) fstype in (ext3) fstype=(ext4) -> /destination,', AppArmorException),
('mount fstype in (ext3) options=(ro) fstype=ext4 -> /destination,', AppArmorException),
) )
def _run_test(self, rawrule, expected): def _run_test(self, rawrule, expected):
@ -198,6 +203,12 @@ class MountTestParseInvalid(AATest):
with self.assertRaises(AppArmorException): with self.assertRaises(AppArmorException):
MountRule('mount', ('=', ('ext4')), ('=', ('bind')), MountRule.ALL, '/foo') MountRule('mount', ('=', ('ext4')), ('=', ('bind')), MountRule.ALL, '/foo')
def test_invalid_cond_type(self):
# cond_type cannot be None if all_values is False
with self.assertRaises(AppArmorBug):
obj = MountConditional('name', [], False, '=', cond_type=None)
obj.is_covered(obj)
class MountTestGlob(AATest): class MountTestGlob(AATest):
def test_glob(self): def test_glob(self):
@ -333,19 +344,24 @@ class MountIsCoveredTest(AATest):
test_obj = MountRule('mount', ('=', ('ext3', 'ext4')), [('in', ('ro')), ('=', ('nosuid', 'noexec'))], '/foo/b*', '/b*') test_obj = MountRule('mount', ('=', ('ext3', 'ext4')), [('in', ('ro')), ('=', ('nosuid', 'noexec'))], '/foo/b*', '/b*')
self.assertTrue(obj.is_equal(test_obj, True), '\n' + test_obj.get_clean() + '\n should be equal to\n' + obj.get_clean()) self.assertTrue(obj.is_equal(test_obj, True), '\n' + test_obj.get_clean() + '\n should be equal to\n' + obj.get_clean())
def test_eq_cond_diff_type(self):
obj = MountConditional('options', ['ro'], False, 'in', 'aare')
test_obj = MountConditional('options', ['ro'], False, 'in', 'list')
self.assertFalse(test_obj == obj, 'cond_types should be different')
def test_eq_cond_diff_instance(self): def test_eq_cond_diff_instance(self):
obj = MountRule('mount', ('=', ('ext3', 'ext4')), [('in', ('ro')), ('=', ('nosuid', 'noexec'))], '/foo/b*', '/b*') obj = MountRule('mount', ('=', ('ext3', 'ext4')), [('in', ('ro')), ('=', ('nosuid', 'noexec'))], '/foo/b*', '/b*')
test_obj = MountConditional('options', ['ro'], False, 'in') test_obj = MountConditional('options', ['ro'], False, 'in', 'list')
self.assertFalse(test_obj == obj, 'Incompatible comparison') self.assertFalse(test_obj == obj, 'Incompatible comparison')
def test_covered_cond_diff_instance(self): def test_covered_cond_diff_instance(self):
obj = MountRule('mount', ('=', ('ext3', 'ext4')), [('in', ('ro')), ('=', ('nosuid', 'noexec'))], '/foo/b*', '/b*') obj = MountRule('mount', ('=', ('ext3', 'ext4')), [('in', ('ro')), ('=', ('nosuid', 'noexec'))], '/foo/b*', '/b*')
test_obj = MountConditional('options', ['ro'], False, 'in') test_obj = MountConditional('options', ['ro'], False, 'in', 'list')
self.assertFalse(test_obj.is_covered(obj), 'Incompatible comparison') self.assertFalse(test_obj.is_covered(obj), 'Incompatible comparison')
def test_covered_cond_diff_name(self): def test_covered_cond_diff_name(self):
obj = MountConditional('foo', ['ro'], False, '=') obj = MountConditional('foo', ['ro'], False, '=', 'list')
test_obj = MountConditional('options', ['ro'], False, '=') test_obj = MountConditional('options', ['ro'], False, '=', 'list')
self.assertFalse(test_obj.is_covered(obj), 'Incompatible comparison') self.assertFalse(test_obj.is_covered(obj), 'Incompatible comparison')
def test_is_notcovered(self): def test_is_notcovered(self):

View File

@ -440,10 +440,6 @@ syntax_failure = (
'file/priority/ok_quoted_4.sd', # quoted string including \" 'file/priority/ok_quoted_4.sd', # quoted string including \"
'file/priority/ok_embedded_spaces_4.sd', # \-escaped space 'file/priority/ok_embedded_spaces_4.sd', # \-escaped space
# mount rules with multiple 'fstype' are not supported by the tools yet, and when writing them, only the last 'fstype' would survive.
# Therefore MountRule intentionally raises an exception when parsing such a rule.
'mount/ok_opt_88.sd', # multiple fstype
# misc # misc
'vars/vars_dbus_12.sd', # AARE starting with {{ are not handled 'vars/vars_dbus_12.sd', # AARE starting with {{ are not handled
'vars/vars_simple_assignment_12.sd', # Redefining existing variable @{BAR} ('\' not handled) 'vars/vars_simple_assignment_12.sd', # Redefining existing variable @{BAR} ('\' not handled)