2
0
mirror of https://gitlab.com/apparmor/apparmor synced 2025-08-22 01:57:43 +00:00

utils: add support for priority rule prefix

Add basic support for the priority rules prefix. This patch does not
allow the utils to set or suggest priorities. It allows parsing and
retaining of the priority prefix if it already exists on rules and
checking if it's in the supported range.

Signed-off-by: Georgia Garcia <georgia.garcia@canonical.com>
Signed-off-by: John Johansen <john.johansen@canonical.com>
This commit is contained in:
John Johansen 2025-02-09 04:35:52 -08:00 committed by Georgia Garcia
parent 3389230437
commit c0fcd1698b
45 changed files with 501 additions and 186 deletions

View File

@ -21,7 +21,7 @@ from apparmor.translations import init_translation
_ = init_translation()
# Profile parsing Regex
RE_AUDIT_DENY = r'^\s*(?P<audit>audit\s+)?(?P<allow>allow\s+|deny\s+)?' # line start, optionally: leading whitespace, <audit> and <allow>/deny
RE_PRIORITY_AUDIT_DENY = r'^\s*(priority\s*=\s*(?P<priority>[+-]?[0-9]*)\s+)?(?P<audit>audit\s+)?(?P<allow>allow\s+|deny\s+)?' # line start, optionally: leading whitespace, <priority> = number, <audit> and <allow>/deny
RE_EOL = r'\s*(?P<comment>#.*?)?\s*$' # optional whitespace, optional <comment>, optional whitespace, end of the line
RE_COMMA_EOL = r'\s*,' + RE_EOL # optional whitespace, comma + RE_EOL
@ -34,8 +34,8 @@ RE_XATTRS = r'(\s+xattrs\s*=\s*\((?P<xattrs>([^)=]+(=[^)=]+)?\s?)+)\)\s*)?'
RE_FLAGS = r'(\s+(flags\s*=\s*)?\((?P<flags>[^)]+)\))?'
RE_PROFILE_END = re.compile(r'^\s*\}' + RE_EOL)
RE_PROFILE_ALL = re.compile(RE_AUDIT_DENY + r'all' + RE_COMMA_EOL)
RE_PROFILE_CAP = re.compile(RE_AUDIT_DENY + r'capability(?P<capability>(\s+\S+)+)?' + RE_COMMA_EOL)
RE_PROFILE_ALL = re.compile(RE_PRIORITY_AUDIT_DENY + r'all' + RE_COMMA_EOL)
RE_PROFILE_CAP = re.compile(RE_PRIORITY_AUDIT_DENY + r'capability(?P<capability>(\s+\S+)+)?' + RE_COMMA_EOL)
RE_PROFILE_ALIAS = re.compile(r'^\s*alias\s+(?P<orig_path>"??.+?"??)\s+->\s*(?P<target>"??.+?"??)' + RE_COMMA_EOL)
RE_PROFILE_RLIMIT = re.compile(r'^\s*set\s+rlimit\s+(?P<rlimit>[a-z]+)\s*<=\s*(?P<value>[^ ]+(\s+[a-zA-Z]+)?)' + RE_COMMA_EOL)
RE_PROFILE_BOOLEAN = re.compile(r'^\s*(?P<varname>\$\{?\w*\}?)\s*=\s*(?P<value>true|false)\s*,?' + RE_EOL, flags=re.IGNORECASE)
@ -43,18 +43,18 @@ RE_PROFILE_VARIABLE = re.compile(r'^\s*(?P<varname>@\{?\w+\}?)\s*(?P<mode>\+?=)\
RE_PROFILE_CONDITIONAL = re.compile(r'^\s*if\s+(not\s+)?(\$\{?\w*\}?)\s*\{' + RE_EOL)
RE_PROFILE_CONDITIONAL_VARIABLE = re.compile(r'^\s*if\s+(not\s+)?defined\s+(@\{?\w+\}?)\s*\{\s*(#.*)?$')
RE_PROFILE_CONDITIONAL_BOOLEAN = re.compile(r'^\s*if\s+(not\s+)?defined\s+(\$\{?\w+\}?)\s*\{\s*(#.*)?$')
RE_PROFILE_NETWORK = re.compile(RE_AUDIT_DENY + r'network(?P<details>\s+.*)?' + RE_COMMA_EOL)
RE_PROFILE_NETWORK = re.compile(RE_PRIORITY_AUDIT_DENY + r'network(?P<details>\s+.*)?' + RE_COMMA_EOL)
RE_PROFILE_CHANGE_HAT = re.compile(r'^\s*\^("??.+?"??)' + RE_COMMA_EOL)
RE_PROFILE_HAT_DEF = re.compile(r'^(?P<leadingspace>\s*)(?P<hat_keyword>\^|hat\s+)(?P<hat>"??[^)]+?"??)' + RE_FLAGS + r'\s*\{' + RE_EOL)
RE_PROFILE_DBUS = re.compile(RE_AUDIT_DENY + r'(dbus\s*,|dbus(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_MOUNT = re.compile(RE_AUDIT_DENY + r'((?P<operation>mount|remount|umount|unmount)(?P<details>\s+[^#]*)?\s*,)' + RE_EOL)
RE_PROFILE_SIGNAL = re.compile(RE_AUDIT_DENY + r'(signal\s*,|signal(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_PTRACE = re.compile(RE_AUDIT_DENY + r'(ptrace\s*,|ptrace(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_PIVOT_ROOT = re.compile(RE_AUDIT_DENY + r'(pivot_root\s*,|pivot_root(?P<details>\s+[^#]*),)' + RE_EOL)
RE_PROFILE_UNIX = re.compile(RE_AUDIT_DENY + r'(unix\s*,|unix(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_USERNS = re.compile(RE_AUDIT_DENY + r'(userns\s*,|userns(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_MQUEUE = re.compile(RE_AUDIT_DENY + r'(mqueue\s*,|mqueue(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_IO_URING = re.compile(RE_AUDIT_DENY + r'(io_uring\s*,|io_uring(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_DBUS = re.compile(RE_PRIORITY_AUDIT_DENY + r'(dbus\s*,|dbus(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_MOUNT = re.compile(RE_PRIORITY_AUDIT_DENY + r'((?P<operation>mount|remount|umount|unmount)(?P<details>\s+[^#]*)?\s*,)' + RE_EOL)
RE_PROFILE_SIGNAL = re.compile(RE_PRIORITY_AUDIT_DENY + r'(signal\s*,|signal(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_PTRACE = re.compile(RE_PRIORITY_AUDIT_DENY + r'(ptrace\s*,|ptrace(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_PIVOT_ROOT = re.compile(RE_PRIORITY_AUDIT_DENY + r'(pivot_root\s*,|pivot_root(?P<details>\s+[^#]*),)' + RE_EOL)
RE_PROFILE_UNIX = re.compile(RE_PRIORITY_AUDIT_DENY + r'(unix\s*,|unix(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_USERNS = re.compile(RE_PRIORITY_AUDIT_DENY + r'(userns\s*,|userns(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_MQUEUE = re.compile(RE_PRIORITY_AUDIT_DENY + r'(mqueue\s*,|mqueue(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
RE_PROFILE_IO_URING = re.compile(RE_PRIORITY_AUDIT_DENY + r'(io_uring\s*,|io_uring(?P<details>\s+[^#]*)\s*,)' + RE_EOL)
# match anything that's not " or #, or matching quotes with anything except quotes inside
__re_no_or_quoted_hash = '([^#"]|"[^"]*")*'
@ -81,7 +81,7 @@ RE_PROFILE_START = re.compile(
RE_PROFILE_CHANGE_PROFILE = re.compile(
RE_AUDIT_DENY
RE_PRIORITY_AUDIT_DENY
+ 'change_profile'
+ r'(\s+' + RE_SAFE_OR_UNSAFE + ')?' # optionally exec mode
+ r'(\s+' + RE_PROFILE_PATH_OR_VAR % 'execcond' + ')?' # optionally exec condition
@ -94,7 +94,7 @@ RE_PROFILE_CHANGE_PROFILE = re.compile(
RE_PATH_PERMS = '(?P<%s>[mrwalkPUCpucix]+)'
RE_PROFILE_FILE_ENTRY = re.compile(
RE_AUDIT_DENY
RE_PRIORITY_AUDIT_DENY
+ r'(?P<owner>owner\s+)?' # optionally: <owner>
+ '('
+ '(?P<bare_file>file)' # bare 'file,' # noqa: E131

View File

@ -41,8 +41,10 @@ class BaseRule(metaclass=ABCMeta):
_match_re = None
def __init__(self, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
"""initialize variables needed by all rule types"""
self._store_priority(priority)
self.audit = audit
self.deny = deny
self.allow_keyword = allow_keyword
@ -52,6 +54,21 @@ class BaseRule(metaclass=ABCMeta):
# Set only in the parse() class method
self.raw_rule = None
def _store_priority(self, priority):
if priority is None: # default priority
self.priority = None
return
try:
ipriority = int(priority)
except ValueError:
raise AppArmorException("Invalid value for priority '%s'" % priority)
if ipriority < -1000 or ipriority > 1000:
raise AppArmorException('priority %d out of range' % (ipriority))
self.priority = ipriority
def _aare_or_all(self, rulepart, partname, is_path, log_event, empty_ok=False):
"""checks rulepart and returns
- (AARE, False) if rulepart is a (non-empty) string
@ -233,7 +250,9 @@ class BaseRule(metaclass=ABCMeta):
"""compare if rule_obj == self
Calls _is_equal_localvars() to compare rule-specific variables"""
if self.audit != rule_obj.audit or self.deny != rule_obj.deny:
if (self.priority != rule_obj.priority
or self.audit != rule_obj.audit
or self.deny != rule_obj.deny):
return False
if strict and (
@ -282,6 +301,9 @@ class BaseRule(metaclass=ABCMeta):
headers = []
qualifier = []
if self.priority:
qualifier.append('priority=%s' % self.priority)
if self.audit:
qualifier.append('audit')
@ -318,7 +340,12 @@ class BaseRule(metaclass=ABCMeta):
raise NotImplementedError("'%s' needs to implement store_edit(), but didn't" % (str(self)))
def modifiers_str(self):
"""return the allow/deny and audit keyword as string, including whitespace"""
"""return priority, allow/deny, and audit keyword as string, including whitespace"""
if self.priority is not None:
prioritystr = 'priority=%s ' % self.priority
else:
prioritystr = ''
if self.audit:
auditstr = 'audit '
@ -332,7 +359,7 @@ class BaseRule(metaclass=ABCMeta):
else:
allowstr = ''
return '%s%s' % (auditstr, allowstr)
return '%s%s%s' % (prioritystr, auditstr, allowstr)
def ensure_modifiers_not_supported(self):
if self.audit:
@ -341,6 +368,8 @@ class BaseRule(metaclass=ABCMeta):
raise AppArmorBug('Attempt to initialize %s with deny flag' % self.__class__.__name__)
if self.allow_keyword:
raise AppArmorBug('Attempt to initialize %s with allow keyword' % self.__class__.__name__)
if self.priority is not None:
raise AppArmorBug('Attempt to initialize %s with priority' % self.__class__.__name__)
class BaseRuleset:
@ -573,9 +602,16 @@ def parse_comment(matches):
def parse_modifiers(matches):
"""returns audit, deny, allow_keyword and comment from the matches object
"""returns priority, audit, deny, allow_keyword and comment from the
matches object
- priority is a number or None
- audit, deny and allow_keyword are True/False
- comment is the comment with a leading space"""
priority = None
if matches.group('priority'):
priority = int(matches.group('priority'))
audit = False
if matches.group('audit'):
audit = True
@ -594,7 +630,7 @@ def parse_modifiers(matches):
comment = parse_comment(matches)
return (audit, deny, allow_keyword, comment)
return (priority, audit, deny, allow_keyword, comment)
def quote_if_needed(data):

View File

@ -29,11 +29,11 @@ class AbiRule(IncludeRule):
_match_re = RE_ABI
def __init__(self, path, ifexists, ismagic, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(path, ifexists, ismagic,
audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# abi doesn't support 'if exists'
if ifexists:

View File

@ -27,12 +27,12 @@ class AliasRule(BaseRule):
_match_re = RE_PROFILE_ALIAS
def __init__(self, orig_path, target, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# aliases don't support allow keyword, audit or deny
# aliases don't support priority, allow keyword, audit or deny
self.ensure_modifiers_not_supported()
if not isinstance(orig_path, str):
@ -62,7 +62,7 @@ class AliasRule(BaseRule):
target = strip_quotes(matches.group('target').strip())
return cls(orig_path, target,
audit=False, deny=False, allow_keyword=False, comment=comment)
audit=False, deny=False, allow_keyword=False, comment=comment, priority=None)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -29,10 +29,10 @@ class AllRule(BaseRule):
_match_re = RE_PROFILE_ALL
def __init__(self, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# no localvars -> nothing more to do
@ -40,11 +40,11 @@ class AllRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
return cls(audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment)
comment=comment, priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -28,12 +28,12 @@ class BooleanRule(BaseRule):
_match_re = RE_PROFILE_BOOLEAN
def __init__(self, varname, value, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# boolean variables don't support allow keyword, audit or deny
# boolean variables don't support priority, allow keyword, audit or deny
self.ensure_modifiers_not_supported()
if not isinstance(varname, str):
@ -63,7 +63,7 @@ class BooleanRule(BaseRule):
value = matches.group('value')
return cls(varname, value,
audit=False, deny=False, allow_keyword=False, comment=comment)
audit=False, deny=False, allow_keyword=False, comment=comment, priority=None)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -46,10 +46,10 @@ class CapabilityRule(BaseRule):
_match_re = RE_PROFILE_CAP
def __init__(self, cap_list, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# Because we support having multiple caps in one rule,
# initializer needs to accept a list of caps.
self.all_caps = False
@ -78,7 +78,7 @@ class CapabilityRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
if matches.group('capability'):
capability = matches.group('capability').strip()
@ -88,7 +88,7 @@ class CapabilityRule(BaseRule):
return cls(capability, audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment)
comment=comment, priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -37,11 +37,11 @@ class ChangeProfileRule(BaseRule):
_match_re = RE_PROFILE_CHANGE_PROFILE
def __init__(self, execmode, execcond, targetprofile, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
"""CHANGE_PROFILE RULE = 'change_profile' [ [ EXEC MODE ] EXEC COND ] [ -> PROGRAMCHILD ]"""
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
if execmode:
if execmode != 'safe' and execmode != 'unsafe':
@ -80,7 +80,7 @@ class ChangeProfileRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
execmode = matches.group('execmode')
@ -95,7 +95,7 @@ class ChangeProfileRule(BaseRule):
targetprofile = cls.ALL
return cls(execmode, execcond, targetprofile,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -80,10 +80,10 @@ class DbusRule(BaseRule):
_match_re = RE_PROFILE_DBUS
def __init__(self, access, bus, path, name, interface, member, peername, peerlabel,
audit=False, deny=False, allow_keyword=False, comment='', log_event=None):
audit=False, deny=False, allow_keyword=False, comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
self.access, self.all_access, unknown_items = check_and_split_list(access, access_keywords, self.ALL, type(self).__name__, 'access')
if unknown_items:
@ -112,7 +112,7 @@ class DbusRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -186,7 +186,8 @@ class DbusRule(BaseRule):
peerlabel = cls.ALL
return cls(access, bus, path, name, interface, member, peername, peerlabel,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment,
priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -47,7 +47,7 @@ class FileRule(BaseRule):
_match_re = RE_PROFILE_FILE_ENTRY
def __init__(self, path, perms, exec_perms, target, owner, file_keyword=False, leading_perms=False,
audit=False, deny=False, allow_keyword=False, comment='', log_event=None):
audit=False, deny=False, allow_keyword=False, comment='', log_event=None, priority=None):
"""Initialize object
Parameters:
@ -61,7 +61,7 @@ class FileRule(BaseRule):
"""
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# rulepart partperms is_path log_event
self.path, self.all_paths = self._aare_or_all(path, 'path', True, log_event) # noqa: E221
@ -138,7 +138,7 @@ class FileRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
owner = bool(matches.group('owner'))
@ -183,7 +183,7 @@ class FileRule(BaseRule):
file_keyword = bool(matches.group('file_keyword'))
return cls(path, perms, exec_perms, target, owner, file_keyword, leading_perms,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -28,12 +28,12 @@ class IncludeRule(BaseRule):
_match_re = RE_INCLUDE
def __init__(self, path, ifexists, ismagic, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# include doesn't support allow keyword, audit or deny
# include doesn't support priority, allow keyword, audit or deny
self.ensure_modifiers_not_supported()
if not isinstance(ifexists, bool):
@ -59,7 +59,7 @@ class IncludeRule(BaseRule):
path, ifexists, ismagic = re_match_include_parse(raw_rule, cls.rule_name)
return cls(path, ifexists, ismagic,
audit=False, deny=False, allow_keyword=False, comment=comment)
audit=False, deny=False, allow_keyword=False, comment=comment, priority=None)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -51,12 +51,12 @@ class IOUringRule(BaseRule):
_match_re = RE_PROFILE_IO_URING
def __init__(self, access, label, audit=False, deny=False,
allow_keyword=False, comment='', log_event=None):
allow_keyword=False, comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment,
log_event=log_event)
log_event=log_event, priority=priority)
self.access, self.all_access, unknown_items = check_and_split_list(access, access_keywords, self.ALL, type(self).__name__, 'access')
if unknown_items:
@ -68,7 +68,7 @@ class IOUringRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
'''parse raw_rule and return instance of this class'''
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -96,7 +96,7 @@ class IOUringRule(BaseRule):
label = cls.ALL
return cls(access, label, audit=audit, deny=deny,
allow_keyword=allow_keyword, comment=comment)
allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
'''return rule (in clean/default formatting)'''

View File

@ -112,12 +112,15 @@ class MountRule(BaseRule):
rule_name = 'mount'
_match_re = RE_PROFILE_MOUNT
def __init__(self, operation, fstype, options, source, dest, audit=False, deny=False, allow_keyword=False, comment='', log_event=None):
def __init__(self, operation, fstype, options, source, dest,
audit=False, deny=False, allow_keyword=False,
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment,
log_event=log_event)
log_event=log_event,
priority=priority)
self.operation = operation
@ -164,7 +167,7 @@ class MountRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
'''parse raw_rule and return instance of this class'''
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
operation = matches.group('operation')
@ -223,7 +226,9 @@ class MountRule(BaseRule):
source = cls.ALL
dest = cls.ALL
return cls(operation=operation, fstype=(is_fstype_equal, fstype), options=(is_options_equal, options), source=source, dest=dest, audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
return cls(operation=operation, fstype=(is_fstype_equal, fstype), options=(is_options_equal, options),
source=source, dest=dest, audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment,
priority=priority)
def get_clean(self, depth=0):
space = ' ' * depth

View File

@ -62,12 +62,13 @@ class MessageQueueRule(BaseRule):
def __init__(self, access, mqueue_type, label, mqueue_name,
audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment,
log_event=log_event)
log_event=log_event,
priority=priority)
self.access, self.all_access, unknown_items = check_and_split_list(access, access_keywords, self.ALL, type(self).__name__, 'access')
if unknown_items:
@ -92,7 +93,7 @@ class MessageQueueRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
'''parse raw_rule and return instance of this class'''
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -132,7 +133,7 @@ class MessageQueueRule(BaseRule):
mqueue_name = cls.ALL
return cls(access, mqueue_type, label, mqueue_name,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
'''return rule (in clean/default formatting)'''

View File

@ -92,10 +92,10 @@ class NetworkRule(BaseRule):
_match_re = RE_PROFILE_NETWORK
def __init__(self, accesses, domain, type_or_protocol, local_expr, peer_expr, audit=False, deny=False,
allow_keyword=False, comment='', log_event=None):
allow_keyword=False, comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
if type(local_expr) is tuple:
if accesses is None:
@ -159,7 +159,7 @@ class NetworkRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -191,7 +191,7 @@ class NetworkRule(BaseRule):
peer_expr = cls.ALL
return cls(accesses, domain, type_or_protocol, local_expr, peer_expr,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -47,12 +47,13 @@ class PivotRootRule(BaseRule):
_match_re = RE_PROFILE_PIVOT_ROOT
# PIVOT ROOT RULE = [ QUALIFIERS ] pivot_root [ oldroot=OLD PUT FILEGLOB ] [ NEW ROOT FILEGLOB ] [ -> PROFILE NAME ]
def __init__(self, oldroot, newroot, profile_name, audit=False, deny=False, allow_keyword=False, comment='', log_event=None):
def __init__(self, oldroot, newroot, profile_name, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment,
log_event=log_event)
log_event=log_event, priority=priority)
self.oldroot, self.all_oldroots = self._aare_or_all(oldroot, 'oldroot', True, log_event) # noqa: E221
self.newroot, self.all_newroots = self._aare_or_all(newroot, 'newroot', True, log_event) # noqa: E221
@ -66,7 +67,7 @@ class PivotRootRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
'''parse raw_rule and return instance of this class'''
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -100,7 +101,7 @@ class PivotRootRule(BaseRule):
profile_name = cls.ALL
return cls(oldroot=oldroot, newroot=newroot, profile_name=profile_name,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
space = ' ' * depth

View File

@ -53,10 +53,10 @@ class PtraceRule(BaseRule):
_match_re = RE_PROFILE_PTRACE
def __init__(self, access, peer, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
self.access, self.all_access, unknown_items = check_and_split_list(
access, access_keywords, self.ALL, type(self).__name__, 'access')
@ -69,7 +69,7 @@ class PtraceRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -98,7 +98,7 @@ class PtraceRule(BaseRule):
peer = cls.ALL
return cls(access, peer,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -49,12 +49,12 @@ class RlimitRule(BaseRule):
_match_re = RE_PROFILE_RLIMIT
def __init__(self, rlimit, value, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# rlimit rules don't support allow keyword, audit or deny
# rlimit rules don't support priority, allow keyword, audit or deny
self.ensure_modifiers_not_supported()
if isinstance(rlimit, str):

View File

@ -78,10 +78,10 @@ class SignalRule(BaseRule):
_match_re = RE_PROFILE_SIGNAL
def __init__(self, access, signal, peer, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
self.access, self.all_access, unknown_items = check_and_split_list(
access, access_keywords, self.ALL, type(self).__name__, 'access')
@ -103,7 +103,7 @@ class SignalRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
"""parse raw_rule and return instance of this class"""
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -141,7 +141,7 @@ class SignalRule(BaseRule):
peer = cls.ALL
return cls(access, signal, peer,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -65,12 +65,14 @@ class UnixRule(BaseRule):
rule_name = 'unix'
_match_re = RE_PROFILE_UNIX
def __init__(self, accesses, rule_conds, local_expr, peer_expr, audit=False, deny=False, allow_keyword=False, comment='', log_event=None):
def __init__(self, accesses, rule_conds, local_expr, peer_expr, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment,
log_event=log_event)
log_event=log_event,
priority=priority)
if type(rule_conds) is tuple: # This comes from the logparser, we convert it to dicts
accesses = strip_parenthesis(accesses).replace(',', ' ').split()
@ -96,7 +98,7 @@ class UnixRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
'''parse raw_rule and return instance of this class'''
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -124,7 +126,8 @@ class UnixRule(BaseRule):
local_expr = cls.ALL
peer_expr = cls.ALL
return cls(accesses=accesses, rule_conds=rule_conds, local_expr=local_expr, peer_expr=peer_expr, audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment)
return cls(accesses=accesses, rule_conds=rule_conds, local_expr=local_expr, peer_expr=peer_expr,
audit=audit, deny=deny, allow_keyword=allow_keyword, comment=comment, priority=priority)
def get_clean(self, depth=0):
space = ' ' * depth

View File

@ -44,12 +44,13 @@ class UserNamespaceRule(BaseRule):
_match_re = RE_PROFILE_USERNS
def __init__(self, access, audit=False, deny=False,
allow_keyword=False, comment='', log_event=None):
allow_keyword=False, comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny,
allow_keyword=allow_keyword,
comment=comment,
log_event=log_event)
log_event=log_event,
priority=priority)
self.access, self.all_access, unknown_items = check_and_split_list(access, access_keyword, self.ALL, type(self).__name__, 'access')
if unknown_items:
@ -59,7 +60,7 @@ class UserNamespaceRule(BaseRule):
def _create_instance(cls, raw_rule, matches):
'''parse raw_rule and return instance of this class'''
audit, deny, allow_keyword, comment = parse_modifiers(matches)
priority, audit, deny, allow_keyword, comment = parse_modifiers(matches)
rule_details = ''
if matches.group('details'):
@ -75,7 +76,7 @@ class UserNamespaceRule(BaseRule):
access = cls.ALL
return cls(access, audit=audit, deny=deny,
allow_keyword=allow_keyword, comment=comment)
allow_keyword=allow_keyword, comment=comment, priority=priority)
@staticmethod
def hashlog_from_event(hl, e):

View File

@ -30,12 +30,12 @@ class VariableRule(BaseRule):
_match_re = RE_PROFILE_VARIABLE
def __init__(self, varname, mode, values, audit=False, deny=False, allow_keyword=False,
comment='', log_event=None):
comment='', log_event=None, priority=None):
super().__init__(audit=audit, deny=deny, allow_keyword=allow_keyword,
comment=comment, log_event=log_event)
comment=comment, log_event=log_event, priority=priority)
# variables don't support allow keyword, audit or deny
# variables don't support priority, allow keyword, audit or deny
self.ensure_modifiers_not_supported()
if not isinstance(varname, str):
@ -70,7 +70,7 @@ class VariableRule(BaseRule):
values = separate_vars(matches.group('values'))
return cls(varname, mode, values,
audit=False, deny=False, allow_keyword=False, comment=comment)
audit=False, deny=False, allow_keyword=False, comment=comment, priority=None)
def get_clean(self, depth=0):
"""return rule (in clean/default formatting)"""

View File

@ -30,10 +30,11 @@ exp = namedtuple('exp', ('comment', 'orig_path', 'target'))
class AliasTest(AATest):
def _compare_obj(self, obj, expected):
# aliases don't support the allow, audit or deny keyword
# aliases don't support the allow, audit, deny, or priority keyword
self.assertEqual(False, obj.allow_keyword)
self.assertEqual(False, obj.audit)
self.assertEqual(False, obj.deny)
self.assertEqual(None, obj.priority)
self.assertEqual(expected.orig_path, obj.orig_path)
self.assertEqual(expected.target, obj.target)
@ -114,6 +115,14 @@ class InvalidAliasInit(AATest):
with self.assertRaises(AppArmorBug):
AliasRule('/foo', '/bar', deny=True)
def test_invalid_priority_1(self):
with self.assertRaises(AppArmorBug):
AliasRule('/foo', '/bar', priority=1)
def test_invalid_priority_2(self):
with self.assertRaises(AppArmorBug):
AliasRule('/foo', '/bar', priority=0)
class InvalidAliasTest(AATest):
def _check_invalid_rawrule(self, rawrule, matches_regex=False):

View File

@ -116,6 +116,7 @@ class WriteAllTestAATest(AATest):
(' deny all ,# foo bar', 'deny all, # foo bar'),
(' allow all ,# foo bar', 'allow all, # foo bar'),
(' allow all ,', 'allow all,'),
(' priority = -2 allow all ,', 'priority=-2 allow all,'),
)
def _run_test(self, rawrule, expected):

View File

@ -71,7 +71,7 @@ class TestBaserule(AATest):
self.ValidSubclass.match('foo')
def test_parse_modifiers_invalid(self):
regex = re.compile(r'^\s*(?P<audit>audit\s+)?(?P<allow>allow\s+|deny\s+|invalid\s+)?')
regex = re.compile(r'^\s*(priority\s*=\s*(?P<priority>[+-]?[0-9]*)\s+)?(?P<audit>audit\s+)?(?P<allow>allow\s+|deny\s+|invalid\s+)?')
matches = regex.search('audit invalid ')
with self.assertRaises(AppArmorBug):

View File

@ -30,10 +30,11 @@ exp = namedtuple('exp', ('comment', 'varname', 'value'))
class BooleanTest(AATest):
def _compare_obj(self, obj, expected):
# boolean variables don't support the allow, audit or deny keyword
# boolean variables don't support the allow, audit, deny, or priority keyword
self.assertEqual(False, obj.allow_keyword)
self.assertEqual(False, obj.audit)
self.assertEqual(False, obj.deny)
self.assertEqual(None, obj.priority)
self.assertEqual(expected.varname, obj.varname)
self.assertEqual(expected.value, obj.value)
@ -122,6 +123,14 @@ class InvalidBooleanInit(AATest):
with self.assertRaises(AppArmorBug):
BooleanRule('$foo', 'true', deny=True)
def test_invalid_priority_1(self):
with self.assertRaises(AppArmorBug):
BooleanRule('$foo', 'true', priority=-100)
def test_invalid_priority_2(self):
with self.assertRaises(AppArmorBug):
BooleanRule('$foo', 'true', priority=0)
class InvalidBooleanTest(AATest):
def _check_invalid_rawrule(self, rawrule, matches_regex=False):

View File

@ -314,6 +314,18 @@ class WriteCapabilityTest(AATest):
self.assertEqual(expected, obj.get_clean(2), 'unexpected clean rule')
self.assertEqual(expected, obj.get_raw(2), 'unexpected raw rule')
def test_write_priority_1(self):
self._check_write_rule(' priority = 923 audit capability sys_admin,', 'priority=923 audit capability sys_admin,')
def test_write_priority_2(self):
self._check_write_rule(' priority = 0 audit capability sys_admin,', 'priority=0 audit capability sys_admin,')
def test_write_priority_3(self):
self._check_write_rule(' priority=-12 audit capability sys_admin,', 'priority=-12 audit capability sys_admin,')
def test_write_priority_4(self):
self._check_write_rule(' priority=+99 audit capability sys_admin,', 'priority=99 audit capability sys_admin,')
class CapabilityCoveredTest(AATest):
def _is_covered(self, obj, rule_to_test):

View File

@ -225,6 +225,10 @@ class WriteChangeProfileTestAATest(AATest):
(' allow change_profile -> /bar ,# foo bar', 'allow change_profile -> /bar, # foo bar'),
(' allow change_profile unsafe /** -> /bar ,# foo bar', 'allow change_profile unsafe /** -> /bar, # foo bar'),
(' allow change_profile "/fo o" -> "/b ar",', 'allow change_profile "/fo o" -> "/b ar",'),
(' priority=9 audit deny change_profile /foo -> baz,', 'priority=9 audit deny change_profile /foo -> baz,'),
(' priority = 0 audit deny change_profile /foo -> baz,', 'priority=0 audit deny change_profile /foo -> baz,'),
(' priority=-54 audit deny change_profile /foo -> baz,', 'priority=-54 audit deny change_profile /foo -> baz,'),
(' priority=+42 audit deny change_profile /foo -> baz,', 'priority=42 audit deny change_profile /foo -> baz,'),
)
def _run_test(self, rawrule, expected):

View File

@ -419,6 +419,10 @@ class WriteDbusTest(AATest):
('dbus receive peer=(label=foo),', 'dbus receive peer=(label=foo),'),
('dbus (send receive) peer=(name=/usr/bin/bar),', 'dbus (receive send) peer=(name=/usr/bin/bar),'),
('dbus (, receive ,,, send ,) interface=/sbin/baz,', 'dbus (receive send) interface=/sbin/baz,'), # XXX leading and trailing ',' inside (...) causes error
(' priority=123 deny dbus send interface = ( foo ),', 'priority=123 deny dbus send interface=foo,'),
(' priority=0 deny dbus send interface = ( foo ),', 'priority=0 deny dbus send interface=foo,'),
(' priority=-72 deny dbus send interface = ( foo ),', 'priority=-72 deny dbus send interface=foo,'),
(' priority=+834 deny dbus send interface = ( foo ),', 'priority=834 deny dbus send interface=foo,'),
# XXX add more complex rules
)

View File

@ -415,6 +415,10 @@ class WriteFileTest(AATest):
(' /foo r,', '/foo r,'),
(' /foo lwr,', '/foo rwl,'),
(' /foo Pxrm -> bar,', '/foo mrPx -> bar,'),
(' priority=-1 /foo Pxrm -> bar,', 'priority=-1 /foo mrPx -> bar,'),
(' priority=0 /foo Pxrm -> bar,', 'priority=0 /foo mrPx -> bar,'),
(' priority=343 /foo Pxrm -> bar,', 'priority=343 /foo mrPx -> bar,'),
(' priority=+65 /foo Pxrm -> bar,', 'priority=65 /foo mrPx -> bar,'),
# with leading permissions
(' audit file r /foo,', 'audit file r /foo,'),
@ -432,11 +436,19 @@ class WriteFileTest(AATest):
(' r /foo ,', 'r /foo,'),
(' klwr /foo ,', 'rwlk /foo,'),
(' Pxrm /foo -> bar,', 'mrPx /foo -> bar,'),
(' priority=1 Pxrm /foo -> bar,', 'priority=1 mrPx /foo -> bar,'),
(' priority=0 Pxrm /foo -> bar,', 'priority=0 mrPx /foo -> bar,'),
(' priority=-22 Pxrm /foo -> bar,', 'priority=-22 mrPx /foo -> bar,'),
(' priority=+78 Pxrm /foo -> bar,', 'priority=78 mrPx /foo -> bar,'),
# link rules
(' link /foo -> /bar,', 'link /foo -> /bar,'),
(' audit deny owner link subset /foo -> /bar,', 'audit deny owner link subset /foo -> /bar,'),
(' link subset /foo -> /bar,', 'link subset /foo -> /bar,')
(' link subset /foo -> /bar,', 'link subset /foo -> /bar,'),
(' priority=0 link /foo -> /bar,', 'priority=0 link /foo -> /bar,'),
(' priority=12 link /foo -> /bar,', 'priority=12 link /foo -> /bar,'),
(' priority=-1 link /foo -> /bar,', 'priority=-1 link /foo -> /bar,'),
(' priority=+4 link /foo -> /bar,', 'priority=4 link /foo -> /bar,'),
)
def test_write_manually_1(self):
@ -844,21 +856,23 @@ class FileSeverityTest(AATest):
class FileLogprofHeaderTest(AATest):
tests = (
# log event old perms ALL / owner
(('file,', set(), set()), [ _('Path'), _('ALL'), _('New Mode'), _('ALL')]), # noqa: E201
(('/foo r,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'r']), # noqa: E201
(('file /bar Px -> foo,', set(), set()), [ _('Path'), '/bar', _('New Mode'), 'Px -> foo']), # noqa: E201
(('deny file,', set(), set()), [_('Qualifier'), 'deny', _('Path'), _('ALL'), _('New Mode'), _('ALL')]),
(('allow file /baz rwk,', set(), set()), [_('Qualifier'), 'allow', _('Path'), '/baz', _('New Mode'), 'rwk']),
(('audit file /foo mr,', set(), set()), [_('Qualifier'), 'audit', _('Path'), '/foo', _('New Mode'), 'mr']),
(('audit deny /foo wk,', set(), set()), [_('Qualifier'), 'audit deny', _('Path'), '/foo', _('New Mode'), 'wk']),
(('owner file /foo ix,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'owner ix']), # noqa: E201
(('audit deny file /foo rlx -> /baz,', set(), set()), [_('Qualifier'), 'audit deny', _('Path'), '/foo', _('New Mode'), 'rlx -> /baz']),
(('/foo rw,', set('r'), set()), [ _('Path'), '/foo', _('Old Mode'), _('r'), _('New Mode'), 'rw']), # noqa: E201
(('/foo rw,', set(), set('rw')), [ _('Path'), '/foo', _('Old Mode'), _('owner rw'), _('New Mode'), 'rw']), # noqa: E201
(('/foo mrw,', set('r'), set('k')), [ _('Path'), '/foo', _('Old Mode'), _('r + owner k'), _('New Mode'), 'mrw']), # noqa: E201
(('/foo mrw,', set('r'), set('rk')), [ _('Path'), '/foo', _('Old Mode'), _('r + owner k'), _('New Mode'), 'mrw']), # noqa: E201
(('link /foo -> /bar,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'link -> /bar']), # noqa: E201
(('link subset /foo -> /bar,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'link subset -> /bar']), # noqa: E201
(('file,', set(), set()), [ _('Path'), _('ALL'), _('New Mode'), _('ALL')]), # noqa: E201
(('/foo r,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'r']), # noqa: E201
(('file /bar Px -> foo,', set(), set()), [ _('Path'), '/bar', _('New Mode'), 'Px -> foo']), # noqa: E201
(('deny file,', set(), set()), [_('Qualifier'), 'deny', _('Path'), _('ALL'), _('New Mode'), _('ALL')]),
(('allow file /baz rwk,', set(), set()), [_('Qualifier'), 'allow', _('Path'), '/baz', _('New Mode'), 'rwk']),
(('audit file /foo mr,', set(), set()), [_('Qualifier'), 'audit', _('Path'), '/foo', _('New Mode'), 'mr']),
(('audit deny /foo wk,', set(), set()), [_('Qualifier'), 'audit deny', _('Path'), '/foo', _('New Mode'), 'wk']),
(('priority=1 file,', set(), set()), [_('Qualifier'), 'priority=1', _('Path'), _('ALL'), _('New Mode'), _('ALL')]),
(('priority=-1 audit file /foo mr,', set(), set()), [_('Qualifier'), 'priority=-1 audit', _('Path'), '/foo', _('New Mode'), 'mr']),
(('owner file /foo ix,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'owner ix']), # noqa: E201
(('audit deny file /foo rlx -> /baz,', set(), set()), [_('Qualifier'), 'audit deny', _('Path'), '/foo', _('New Mode'), 'rlx -> /baz']),
(('/foo rw,', set('r'), set()), [ _('Path'), '/foo', _('Old Mode'), _('r'), _('New Mode'), 'rw']), # noqa: E201
(('/foo rw,', set(), set('rw')), [ _('Path'), '/foo', _('Old Mode'), _('owner rw'), _('New Mode'), 'rw']), # noqa: E201
(('/foo mrw,', set('r'), set('k')), [ _('Path'), '/foo', _('Old Mode'), _('r + owner k'), _('New Mode'), 'mrw']), # noqa: E201
(('/foo mrw,', set('r'), set('rk')), [ _('Path'), '/foo', _('Old Mode'), _('r + owner k'), _('New Mode'), 'mrw']), # noqa: E201
(('link /foo -> /bar,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'link -> /bar']), # noqa: E201
(('link subset /foo -> /bar,', set(), set()), [ _('Path'), '/foo', _('New Mode'), 'link subset -> /bar']), # noqa: E201
)
def _run_test(self, params, expected):

View File

@ -37,6 +37,7 @@ class IncludeTest(AATest):
self.assertEqual(False, obj.allow_keyword) # not supported in include rules, expected to be always False
self.assertEqual(False, obj.audit) # not supported in include rules, expected to be always False
self.assertEqual(False, obj.deny) # not supported in include rules, expected to be always False
self.assertEqual(None, obj.priority) # not supported in include rules, expected to be always None
self.assertEqual(expected.comment, obj.comment)
self.assertEqual(expected.path, obj.path)
@ -159,6 +160,10 @@ class InvalidIncludeInit(AATest):
with self.assertRaises(AppArmorBug):
IncludeRule('foo', False, False, deny=True)
def test_priority_true(self):
with self.assertRaises(AppArmorBug):
IncludeRule('foo', False, False, priority=0)
class InvalidIncludeTest(AATest):
def _check_invalid_rawrule(self, rawrule, matches_regex=False):

View File

@ -64,10 +64,10 @@ class IOUringTestParseInvalid(AATest):
IOUringRule.create_instance('foo,')
def test_diff_non_iouringrule(self):
exp = namedtuple('exp', ('audit', 'deny'))
exp = namedtuple('exp', ('audit', 'deny', 'priority'))
obj = IOUringRule(('sqpoll'), IOUringRule.ALL)
with self.assertRaises(AppArmorBug):
obj.is_equal(exp(False, False), False)
obj.is_equal(exp(False, False, None), False)
def test_diff_access(self):
obj1 = IOUringRule(IOUringRule.ALL, IOUringRule.ALL)
@ -124,6 +124,10 @@ class WriteIOUringTestAATest(AATest):
('io_uring sqpoll label="tst",', 'io_uring sqpoll label="tst",'),
('io_uring (override_creds) label=bar,', 'io_uring override_creds label=bar,'),
('io_uring (sqpoll override_creds) label=/foo,', 'io_uring (override_creds sqpoll) label=/foo,'),
(' priority=1 deny io_uring override_creds ,# foo bar', 'priority=1 deny io_uring override_creds, # foo bar'),
(' priority=0 deny io_uring override_creds ,# foo bar', 'priority=0 deny io_uring override_creds, # foo bar'),
(' priority=-23 deny io_uring override_creds ,# foo bar', 'priority=-23 deny io_uring override_creds, # foo bar'),
(' priority=+21 deny io_uring override_creds ,# foo bar', 'priority=21 deny io_uring override_creds, # foo bar'),
)
def _run_test(self, rawrule, expected):

View File

@ -0,0 +1,90 @@
#! /usr/bin/python3
# ------------------------------------------------------------------
#
# Copyright (C) 2025 Canonical Ltd.
#
# This program is free software; you can redistribute it and/or
# modify it under the terms of version 2 of the GNU General Public
# License as published by the Free Software Foundation.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# ----------------------------------------------------------------------
import unittest
from apparmor.common import AppArmorException
from apparmor.rule.capability import CapabilityRule
from apparmor.rule.change_profile import ChangeProfileRule
from apparmor.rule.dbus import DbusRule
from apparmor.rule.file import FileRule
from apparmor.rule.io_uring import IOUringRule
from apparmor.rule.mount import MountRule
from apparmor.rule.mqueue import MessageQueueRule
from apparmor.rule.network import NetworkRule
from apparmor.rule.pivot_root import PivotRootRule
from apparmor.rule.ptrace import PtraceRule
from apparmor.rule.signal import SignalRule
from apparmor.rule.unix import UnixRule
from apparmor.rule.userns import UserNamespaceRule
from apparmor.rule.all import AllRule
from common_test import AATest, setup_all_loops
class TestInvalid_parse_priority(AATest):
tests = (
((CapabilityRule, 'priority=a capability,'), AppArmorException),
((DbusRule, 'priority=a dbus,'), AppArmorException),
((MountRule, 'priority=a mount,'), AppArmorException),
((MountRule, 'priority=a umount,'), AppArmorException),
((MountRule, 'priority=a unmount,'), AppArmorException),
((MountRule, 'priority=a remount,'), AppArmorException),
((SignalRule, 'priority=a signal,'), AppArmorException),
((PtraceRule, 'priority=a ptrace,'), AppArmorException),
((PivotRootRule, 'priority=a pivot_root,'), AppArmorException),
((UnixRule, 'priority=a unix,'), AppArmorException),
((NetworkRule, 'priority=a network,'), AppArmorException),
((UserNamespaceRule, 'priority=a userns,'), AppArmorException),
((MessageQueueRule, 'priority=a mqueue,'), AppArmorException),
((IOUringRule, 'priority=a io_uring,'), AppArmorException),
((ChangeProfileRule, 'priority=a change_profile,'), AppArmorException),
((FileRule, 'priority=a file,'), AppArmorException),
((AllRule, 'priority=a all,'), AppArmorException),
)
def _run_test(self, params, expected):
rule_cls, rule = params
with self.assertRaises(expected):
rule_cls.create_instance(rule) # Invalid rule
class TestInvalid_init_priority(AATest):
tests = (
((CapabilityRule, (CapabilityRule.ALL,)), AppArmorException),
((DbusRule, (DbusRule.ALL,) * 8), AppArmorException),
((MountRule, (MountRule.ALL,) * 5), AppArmorException),
((SignalRule, (SignalRule.ALL,) * 3), AppArmorException),
((PtraceRule, (PtraceRule.ALL,) * 2), AppArmorException),
((PivotRootRule, (PivotRootRule.ALL,) * 3), AppArmorException),
((UnixRule, (UnixRule.ALL,) * 4), AppArmorException),
((NetworkRule, (NetworkRule.ALL,) * 5), AppArmorException),
((UserNamespaceRule, (UserNamespaceRule.ALL,) * 1), AppArmorException),
((MessageQueueRule, (MessageQueueRule.ALL,) * 4), AppArmorException),
((IOUringRule, (IOUringRule.ALL,) * 2), AppArmorException),
((ChangeProfileRule, (ChangeProfileRule.ALL,) * 3), AppArmorException),
((FileRule, (FileRule.ALL,) * 5), AppArmorException),
((AllRule, ()), AppArmorException),
)
def _run_test(self, params, expected):
rule_cls, args = params
with self.assertRaises(expected):
rule_cls(*args, priority="invalid") # ValueError
setup_all_loops(__name__)
if __name__ == '__main__':
unittest.main(verbosity=1)

View File

@ -123,10 +123,10 @@ class MountTestParseInvalid(AATest):
MountRule.create_instance('foo,')
def test_diff_non_mountrule(self):
exp = namedtuple('exp', ('audit', 'deny'))
exp = namedtuple('exp', ('audit', 'deny', 'priority'))
obj = MountRule('mount', ('=', ['ext4']), MountRule.ALL, MountRule.ALL, MountRule.ALL)
with self.assertRaises(AppArmorBug):
obj.is_equal(exp(False, False), False)
obj.is_equal(exp(False, False, None), False)
def test_diff_invalid_fstype_equals_or_in(self):
with self.assertRaises(AppArmorBug):
@ -230,6 +230,12 @@ class MountTestClean(AATest):
(' umount /foo , ', 'umount /foo,'),
(' remount , ', 'remount,'),
(' remount /foo , ', 'remount /foo,'),
('priority =1 mount "" -> /foo , ', 'priority=1 mount "" -> /foo,'),
('priority=0 audit mount "/f /b" -> "/foo bar" , ', 'priority=0 audit mount "/f /b" -> "/foo bar",'),
(' priority = +10 umount , ', 'priority=10 umount,'),
(' priority=-2 deny umount /foo , ', 'priority=-2 deny umount /foo,'),
('priority= 32 audit deny remount , ', 'priority=32 audit deny remount,'),
(' priority = -32 remount /foo , ', 'priority=-32 remount /foo,'),
)
def _run_test(self, rawrule, expected):

View File

@ -77,10 +77,10 @@ class MessageQueueTestParseInvalid(AATest):
MessageQueueRule.create_instance('foo,')
def test_diff_non_mqueuerule(self):
exp = namedtuple('exp', ('audit', 'deny'))
exp = namedtuple('exp', ('audit', 'deny', 'priority'))
obj = MessageQueueRule(('open'), 'posix', 'bar', '/foo')
with self.assertRaises(AppArmorBug):
obj.is_equal(exp(False, False), False)
obj.is_equal(exp(False, False, None), False)
def test_diff_access(self):
obj1 = MessageQueueRule(('open'), 'posix', 'bar', '/foo')
@ -171,6 +171,10 @@ class WriteMessageQueueTestAATest(AATest):
('mqueue wr label=tst 1234,', 'mqueue wr label=tst 1234,'),
('mqueue wr type=sysv label=tst 1234,', 'mqueue wr type=sysv label=tst 1234,'),
('mqueue wr type=posix label=tst /foo,', 'mqueue wr type=posix label=tst /foo,'),
(' priority = -82 mqueue getattr /foo,', 'priority=-82 mqueue getattr /foo,'),
(' priority = 12 audit mqueue (setattr getattr) 1234,', 'priority=12 audit mqueue (getattr setattr) 1234,'),
(' priority=0 mqueue getattr /foo,', 'priority=0 mqueue getattr /foo,'),
(' priority=+82 mqueue getattr /foo,', 'priority=82 mqueue getattr /foo,'),
)
def _run_test(self, rawrule, expected):

View File

@ -286,6 +286,10 @@ class WriteNetworkTestAATest(AATest):
(' network stream peer = ( ip=::1 port=22 ) ,', 'network stream peer=(ip=::1 port=22),'),
(' network ( bind , listen ) stream ip = ::1 port = 22 ,', 'network (bind, listen) stream ip=::1 port=22,'),
(' allow network tcp ,# foo bar', 'allow network tcp, # foo bar'),
(' priority = -02 allow network tcp ,# foo bar', 'priority=-2 allow network tcp, # foo bar'),
(' priority = 0 allow network tcp ,# foo bar', 'priority=0 allow network tcp, # foo bar'),
(' priority = 43 allow network tcp ,# foo bar', 'priority=43 allow network tcp, # foo bar'),
(' priority=+123 allow network tcp ,# foo bar', 'priority=123 allow network tcp, # foo bar'),
)

View File

@ -35,9 +35,6 @@ skip_startswith = (
# Pux and Cux (which actually mean PUx and CUx) get rejected by the tools
'generated_x/exact-',
# don't handle rule priorities yet
'file/priority/',
)
# testcases that should raise an exception, but don't
@ -246,11 +243,16 @@ unknown_line = (
'file/ok_other_1.sd',
'file/ok_other_2.sd',
'file/ok_other_3.sd',
'file/priority/ok_other_1.sd',
'file/priority/ok_other_2.sd',
'file/priority/ok_other_3.sd',
# 'unsafe' keyword
'file/file/front_perms_ok_2.sd',
'file/front_perms_ok_2.sd',
'xtrans/simple_ok_cx_1.sd',
'file/priority/front_perms_ok_1.sd',
'file/priority/front_perms_ok_2.sd',
# owner / audit {...} blocks
'file/file/owner/ok_1.sd',
@ -355,6 +357,9 @@ syntax_failure = (
'file/ok_5.sd', # Invalid mode UX
'file/ok_2.sd', # Invalid mode RWM
'file/ok_4.sd', # Invalid mode iX
'file/priority/ok_5.sd', # Invalid mode UX
'file/priority/ok_2.sd', # Invalid mode RWM
'file/priority/ok_4.sd', # Invalid mode iX
'xtrans/simple_ok_pix_1.sd', # Invalid mode pIx
'xtrans/simple_ok_pux_1.sd', # Invalid mode rPux
@ -424,6 +429,8 @@ syntax_failure = (
'file/ok_embedded_spaces_4.sd', # \-escaped space
'file/file/ok_embedded_spaces_4.sd', # \-escaped space
'file/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
# mount rules with multiple 'options' or 'fstype' are not supported by the tools yet, and when writing them, only the last 'options'/'fstype' would survive.
# Therefore MountRule intentionally raises an exception when parsing such a rule.

View File

@ -261,6 +261,10 @@ class WritePivotRootTestAATest(AATest):
(' pivot_root oldroot="/old" , # foo ', 'pivot_root oldroot=/old, # foo'),
(' pivot_root oldroot=/old -> some_profile , ', 'pivot_root oldroot=/old -> some_profile,'),
(' pivot_root oldroot=/old /new -> some_profile , ', 'pivot_root oldroot=/old /new -> some_profile,'),
('priority=1 pivot_root oldroot=/old /new -> some_profile , ', 'priority=1 pivot_root oldroot=/old /new -> some_profile,'),
('priority=0 pivot_root oldroot=/old /new -> some_profile , ', 'priority=0 pivot_root oldroot=/old /new -> some_profile,'),
('priority=-1 pivot_root oldroot=/old /new -> some_profile , ', 'priority=-1 pivot_root oldroot=/old /new -> some_profile,'),
('priority=+1 pivot_root oldroot=/old /new -> some_profile , ', 'priority=1 pivot_root oldroot=/old /new -> some_profile,'),
)
def test_write_manually(self):

View File

@ -260,6 +260,10 @@ class WritePtraceTestAATest(AATest):
('ptrace (read tracedby) peer=/usr/bin/bar,', 'ptrace (read tracedby) peer=/usr/bin/bar,'),
('ptrace (trace read) peer=/usr/bin/bar,', 'ptrace (read trace) peer=/usr/bin/bar,'),
('ptrace wr peer=/sbin/baz,', 'ptrace wr peer=/sbin/baz,'),
('priority = 010 deny ptrace ( read ), ', 'priority=10 deny ptrace read,'),
('priority = 0 deny ptrace ( read ), ', 'priority=0 deny ptrace read,'),
('priority = -21 deny ptrace ( read ), ', 'priority=-21 deny ptrace read,'),
('priority = +83 deny ptrace ( read ), ', 'priority=83 deny ptrace read,'),
)
def test_write_manually(self):

View File

@ -218,11 +218,16 @@ class AARegexCapability(AARegexTest):
self.regex = RE_PROFILE_CAP
tests = (
(' capability net_raw,', (None, None, 'net_raw', 'net_raw', None)),
('capability net_raw , ', (None, None, 'net_raw', 'net_raw', None)),
(' capability,', (None, None, None, None, None)),
(' capability , ', (None, None, None, None, None)),
(' capabilitynet_raw,', False)
(' capability net_raw,', (None, None, None, None, 'net_raw', 'net_raw', None)),
('capability net_raw , ', (None, None, None, None, 'net_raw', 'net_raw', None)),
(' capability,', (None, None, None, None, None, None, None)),
(' capability , ', (None, None, None, None, None, None, None)),
(' capabilitynet_raw,', False),
(' priority=1 capability net_raw,', ('priority=1', '1', None, None, 'net_raw', 'net_raw', None)),
('priority=1 capability net_raw , ', ('priority=1', '1', None, None, 'net_raw', 'net_raw', None)),
(' priority=1 capability,', ('priority=1', '1', None, None, None, None, None)),
(' priority=1 capability , ', ('priority=1', '1', None, None, None, None, None)),
(' priority=1 capabilitynet_raw,', False),
)
@ -233,13 +238,19 @@ class AARegexDbus(AARegexTest):
self.regex = RE_PROFILE_DBUS
tests = (
(' dbus,', (None, None, 'dbus,', None, None)),
(' audit dbus,', ('audit', None, 'dbus,', None, None)),
(' dbus send member=no_comment,', (None, None, 'dbus send member=no_comment,', 'send member=no_comment', None)),
(' dbus send member=no_comment, # comment', (None, None, 'dbus send member=no_comment,', 'send member=no_comment', '# comment')),
(' dbus,', (None, None, None, None, 'dbus,', None, None)),
(' audit dbus,', (None, None, 'audit', None, 'dbus,', None, None)),
(' dbus send member=no_comment,', (None, None, None, None, 'dbus send member=no_comment,', 'send member=no_comment', None)),
(' dbus send member=no_comment, # comment', (None, None, None, None, 'dbus send member=no_comment,', 'send member=no_comment', '# comment')),
(' priority=-11 dbus,', ('priority=-11', '-11', None, None, 'dbus,', None, None)),
(' priority=-11 audit dbus,', ('priority=-11', '-11', 'audit', None, 'dbus,', None, None)),
(' priority=-11 dbus send member=no_comment,', ('priority=-11', '-11', None, None, 'dbus send member=no_comment,', 'send member=no_comment', None)),
(' priority=-11 dbus send member=no_comment, # comment', ('priority=-11', '-11', None, None, 'dbus send member=no_comment,', 'send member=no_comment', '# comment')),
(' dbusdriver,', False),
(' audit dbusdriver,', False),
(' priority=-11 audit dbusdriver,', False),
)
@ -250,19 +261,30 @@ class AARegexMount(AARegexTest):
self.regex = RE_PROFILE_MOUNT
tests = (
(' mount,', (None, None, 'mount,', 'mount', None, None)),
(' audit mount,', ('audit', None, 'mount,', 'mount', None, None)),
(' umount,', (None, None, 'umount,', 'umount', None, None)),
(' audit umount,', ('audit', None, 'umount,', 'umount', None, None)),
(' unmount,', (None, None, 'unmount,', 'unmount', None, None)),
(' audit unmount,', ('audit', None, 'unmount,', 'unmount', None, None)),
(' remount,', (None, None, 'remount,', 'remount', None, None)),
(' deny remount,', (None, 'deny', 'remount,', 'remount', None, None)),
(' mount,', (None, None, None, None, 'mount,', 'mount', None, None)),
(' audit mount,', (None, None, 'audit', None, 'mount,', 'mount', None, None)),
(' umount,', (None, None, None, None, 'umount,', 'umount', None, None)),
(' audit umount,', (None, None, 'audit', None, 'umount,', 'umount', None, None)),
(' unmount,', (None, None, None, None, 'unmount,', 'unmount', None, None)),
(' audit unmount,', (None, None, 'audit', None, 'unmount,', 'unmount', None, None)),
(' remount,', (None, None, None, None, 'remount,', 'remount', None, None)),
(' deny remount,', (None, None, None, 'deny', 'remount,', 'remount', None, None)),
(' mount, # comment', (None, None, 'mount,', 'mount', None, '# comment')),
(' priority = 0 mount,', ('priority = 0', '0', None, None, 'mount,', 'mount', None, None)),
(' priority = 0 audit mount,', ('priority = 0', '0', 'audit', None, 'mount,', 'mount', None, None)),
(' priority = 0 umount,', ('priority = 0', '0', None, None, 'umount,', 'umount', None, None)),
(' priority = 0 audit umount,', ('priority = 0', '0', 'audit', None, 'umount,', 'umount', None, None)),
(' priority = 0 unmount,', ('priority = 0', '0', None, None, 'unmount,', 'unmount', None, None)),
(' priority = 0 audit unmount,', ('priority = 0', '0', 'audit', None, 'unmount,', 'unmount', None, None)),
(' priority = 0 remount,', ('priority = 0', '0', None, None, 'remount,', 'remount', None, None)),
(' priority = 0 deny remount,', ('priority = 0', '0', None, 'deny', 'remount,', 'remount', None, None)),
(' mount, # comment', (None, None, None, None, 'mount,', 'mount', None, '# comment')),
(' priority = 0 mount, # comment', ('priority = 0', '0', None, None, 'mount,', 'mount', None, '# comment')),
(' mountain,', False),
(' audit mountain,', False),
(' priority = 0 audit mountain,', False),
)
@ -273,16 +295,25 @@ class AARegexSignal(AARegexTest):
self.regex = RE_PROFILE_SIGNAL
tests = (
(' signal,', (None, None, 'signal,', None, None)),
(' audit signal,', ('audit', None, 'signal,', None, None)),
(' signal receive,', (None, None, 'signal receive,', 'receive', None)),
(' signal (send, receive),', (None, None, 'signal (send, receive),', '(send, receive)', None)),
(' audit signal (receive),', ('audit', None, 'signal (receive),', '(receive)', None)),
(' signal (send, receive) set=(usr1 usr2),', (None, None, 'signal (send, receive) set=(usr1 usr2),', '(send, receive) set=(usr1 usr2)', None)),
(' signal send set=(hup, quit) peer=/usr/sbin/daemon,', (None, None, 'signal send set=(hup, quit) peer=/usr/sbin/daemon,', 'send set=(hup, quit) peer=/usr/sbin/daemon', None)),
(' signal,', (None, None, None, None, 'signal,', None, None)),
(' audit signal,', (None, None, 'audit', None, 'signal,', None, None)),
(' signal receive,', (None, None, None, None, 'signal receive,', 'receive', None)),
(' signal (send, receive),', (None, None, None, None, 'signal (send, receive),', '(send, receive)', None)),
(' audit signal (receive),', (None, None, 'audit', None, 'signal (receive),', '(receive)', None)),
(' signal (send, receive) set=(usr1 usr2),', (None, None, None, None, 'signal (send, receive) set=(usr1 usr2),', '(send, receive) set=(usr1 usr2)', None)),
(' signal send set=(hup, quit) peer=/usr/sbin/daemon,', (None, None, None, None, 'signal send set=(hup, quit) peer=/usr/sbin/daemon,', 'send set=(hup, quit) peer=/usr/sbin/daemon', None)),
(' priority = -1 signal,', ('priority = -1', '-1', None, None, 'signal,', None, None)),
(' priority = -1 audit signal,', ('priority = -1', '-1', 'audit', None, 'signal,', None, None)),
(' priority = -1 signal receive,', ('priority = -1', '-1', None, None, 'signal receive,', 'receive', None)),
(' priority = -1 signal (send, receive),', ('priority = -1', '-1', None, None, 'signal (send, receive),', '(send, receive)', None)),
(' priority = -1 audit signal (receive),', ('priority = -1', '-1', 'audit', None, 'signal (receive),', '(receive)', None)),
(' priority = -1 signal (send, receive) set=(usr1 usr2),', ('priority = -1', '-1', None, None, 'signal (send, receive) set=(usr1 usr2),', '(send, receive) set=(usr1 usr2)', None)),
(' priority = -1 signal send set=(hup, quit) peer=/usr/sbin/daemon,', ('priority = -1', '-1', None, None, 'signal send set=(hup, quit) peer=/usr/sbin/daemon,', 'send set=(hup, quit) peer=/usr/sbin/daemon', None)),
(' signalling,', False),
(' audit signalling,', False),
(' priority = -1 audit signalling,', False),
(' signalling receive,', False),
)
@ -294,16 +325,24 @@ class AARegexPtrace(AARegexTest):
self.regex = RE_PROFILE_PTRACE
tests = (
# audit allow rule rule details comment
(' ptrace,', (None, None, 'ptrace,', None, None)),
(' audit ptrace,', ('audit', None, 'ptrace,', None, None)),
(' ptrace trace,', (None, None, 'ptrace trace,', 'trace', None)),
(' ptrace (tracedby, readby),', (None, None, 'ptrace (tracedby, readby),', '(tracedby, readby)', None)),
(' audit ptrace (read),', ('audit', None, 'ptrace (read),', '(read)', None)),
(' ptrace trace peer=/usr/sbin/daemon,', (None, None, 'ptrace trace peer=/usr/sbin/daemon,', 'trace peer=/usr/sbin/daemon', None)),
# priority audit allow rule rule details comment
(' ptrace,', (None, None, None, None, 'ptrace,', None, None)),
(' audit ptrace,', (None, None, 'audit', None, 'ptrace,', None, None)),
(' ptrace trace,', (None, None, None, None, 'ptrace trace,', 'trace', None)),
(' ptrace (tracedby, readby),', (None, None, None, None, 'ptrace (tracedby, readby),', '(tracedby, readby)', None)),
(' audit ptrace (read),', (None, None, 'audit', None, 'ptrace (read),', '(read)', None)),
(' ptrace trace peer=/usr/sbin/daemon,', (None, None, None, None, 'ptrace trace peer=/usr/sbin/daemon,', 'trace peer=/usr/sbin/daemon', None)),
(' priority=100 ptrace,', ('priority=100', '100', None, None, 'ptrace,', None, None)),
(' priority=100 audit ptrace,', ('priority=100', '100', 'audit', None, 'ptrace,', None, None)),
(' priority=100 ptrace trace,', ('priority=100', '100', None, None, 'ptrace trace,', 'trace', None)),
(' priority=100 ptrace (tracedby, readby),', ('priority=100', '100', None, None, 'ptrace (tracedby, readby),', '(tracedby, readby)', None)),
(' priority=100 audit ptrace (read),', ('priority=100', '100', 'audit', None, 'ptrace (read),', '(read)', None)),
(' priority=100 ptrace trace peer=/usr/sbin/daemon,', ('priority=100', '100', None, None, 'ptrace trace peer=/usr/sbin/daemon,', 'trace peer=/usr/sbin/daemon', None)),
(' ptraceback,', False),
(' audit ptraceback,', False),
(' priority=100 audit ptraceback,', False),
(' ptraceback trace,', False),
)
@ -315,12 +354,19 @@ class AARegexPivotRoot(AARegexTest):
self.regex = RE_PROFILE_PIVOT_ROOT
tests = (
(' pivot_root,', (None, None, 'pivot_root,', None, None)),
(' audit pivot_root,', ('audit', None, 'pivot_root,', None, None)),
(' pivot_root oldroot=/new/old,', (None, None, 'pivot_root oldroot=/new/old,', 'oldroot=/new/old', None)),
(' pivot_root oldroot=/new/old /new,', (None, None, 'pivot_root oldroot=/new/old /new,', 'oldroot=/new/old /new', None)),
(' pivot_root oldroot=/new/old /new -> child,', (None, None, 'pivot_root oldroot=/new/old /new -> child,', 'oldroot=/new/old /new -> child', None)),
(' audit pivot_root oldroot=/new/old /new -> child,', ('audit', None, 'pivot_root oldroot=/new/old /new -> child,', 'oldroot=/new/old /new -> child', None)),
(' pivot_root,', (None, None, None, None, 'pivot_root,', None, None)),
(' audit pivot_root,', (None, None, 'audit', None, 'pivot_root,', None, None)),
(' pivot_root oldroot=/new/old,', (None, None, None, None, 'pivot_root oldroot=/new/old,', 'oldroot=/new/old', None)),
(' pivot_root oldroot=/new/old /new,', (None, None, None, None, 'pivot_root oldroot=/new/old /new,', 'oldroot=/new/old /new', None)),
(' pivot_root oldroot=/new/old /new -> child,', (None, None, None, None, 'pivot_root oldroot=/new/old /new -> child,', 'oldroot=/new/old /new -> child', None)),
(' audit pivot_root oldroot=/new/old /new -> child,', (None, None, 'audit', None, 'pivot_root oldroot=/new/old /new -> child,', 'oldroot=/new/old /new -> child', None)),
(' priority=-100 pivot_root,', ('priority=-100', '-100', None, None, 'pivot_root,', None, None)),
(' priority=-100 audit pivot_root,', ('priority=-100', '-100', 'audit', None, 'pivot_root,', None, None)),
(' priority=-100 pivot_root oldroot=/new/old,', ('priority=-100', '-100', None, None, 'pivot_root oldroot=/new/old,', 'oldroot=/new/old', None)),
(' priority=-100 pivot_root oldroot=/new/old /new,', ('priority=-100', '-100', None, None, 'pivot_root oldroot=/new/old /new,', 'oldroot=/new/old /new', None)),
(' priority=-100 pivot_root oldroot=/new/old /new -> child,', ('priority=-100', '-100', None, None, 'pivot_root oldroot=/new/old /new -> child,', 'oldroot=/new/old /new -> child', None)),
(' priority=-100 audit pivot_root oldroot=/new/old /new -> child,', ('priority=-100', '-100', 'audit', None, 'pivot_root oldroot=/new/old /new -> child,', 'oldroot=/new/old /new -> child', None)),
('pivot_root', False), # comma missing
@ -329,6 +375,7 @@ class AARegexPivotRoot(AARegexTest):
('pivot_rootbeer, # comment', False),
('pivot_rootbeer /new, ', False),
('pivot_rootbeer /new, # comment', False),
('priority=-100 pivot_rootbeer /new, # comment', False),
)
@ -339,20 +386,35 @@ class AARegexUnix(AARegexTest):
self.regex = RE_PROFILE_UNIX
tests = (
(' unix,', (None, None, 'unix,', None, None)),
(' audit unix,', ('audit', None, 'unix,', None, None)),
(' unix accept,', (None, None, 'unix accept,', 'accept', None)),
(' allow unix connect,', (None, 'allow', 'unix connect,', 'connect', None)),
(' audit allow unix bind,', ('audit', 'allow', 'unix bind,', 'bind', None)),
(' deny unix bind,', (None, 'deny', 'unix bind,', 'bind', None)),
('unix peer=(label=@{profile_name}),', (None, None, 'unix peer=(label=@{profile_name}),', 'peer=(label=@{profile_name})', None)),
('unix (receive) peer=(label=unconfined),', (None, None, 'unix (receive) peer=(label=unconfined),', '(receive) peer=(label=unconfined)', None)),
(' unix (getattr, shutdown) peer=(addr=none),', (None, None, 'unix (getattr, shutdown) peer=(addr=none),', '(getattr, shutdown) peer=(addr=none)', None)),
('unix (connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*"),', (None, None, 'unix (connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*"),',
(' unix,', (None, None, None, None, 'unix,', None, None)),
(' audit unix,', (None, None, 'audit', None, 'unix,', None, None)),
(' unix accept,', (None, None, None, None, 'unix accept,', 'accept', None)),
(' allow unix connect,', (None, None, None, 'allow', 'unix connect,', 'connect', None)),
(' audit allow unix bind,', (None, None, 'audit', 'allow', 'unix bind,', 'bind', None)),
(' deny unix bind,', (None, None, None, 'deny', 'unix bind,', 'bind', None)),
('unix peer=(label=@{profile_name}),', (None, None, None, None, 'unix peer=(label=@{profile_name}),', 'peer=(label=@{profile_name})', None)),
('unix (receive) peer=(label=unconfined),', (None, None, None, None, 'unix (receive) peer=(label=unconfined),', '(receive) peer=(label=unconfined)', None)),
(' unix (getattr, shutdown) peer=(addr=none),', (None, None, None, None, 'unix (getattr, shutdown) peer=(addr=none),', '(getattr, shutdown) peer=(addr=none)', None)),
('unix (connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*"),', (None, None, None, None, 'unix (connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*"),',
'(connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*")', # noqa: E127
None)), # noqa: E127
(' priority=1 unix,', ('priority=1', '1', None, None, 'unix,', None, None)),
(' priority=1 audit unix,', ('priority=1', '1', 'audit', None, 'unix,', None, None)),
(' priority=1 unix accept,', ('priority=1', '1', None, None, 'unix accept,', 'accept', None)),
(' priority=1 allow unix connect,', ('priority=1', '1', None, 'allow', 'unix connect,', 'connect', None)),
(' priority=1 audit allow unix bind,', ('priority=1', '1', 'audit', 'allow', 'unix bind,', 'bind', None)),
(' priority=1 deny unix bind,', ('priority=1', '1', None, 'deny', 'unix bind,', 'bind', None)),
('priority=1 unix peer=(label=@{profile_name}),', ('priority=1', '1', None, None, 'unix peer=(label=@{profile_name}),', 'peer=(label=@{profile_name})', None)),
('priority=1 unix (receive) peer=(label=unconfined),', ('priority=1', '1', None, None, 'unix (receive) peer=(label=unconfined),', '(receive) peer=(label=unconfined)', None)),
(' priority=1 unix (getattr, shutdown) peer=(addr=none),', ('priority=1', '1', None, None, 'unix (getattr, shutdown) peer=(addr=none),', '(getattr, shutdown) peer=(addr=none)', None)),
('priority=1 unix (connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*"),', ('priority=1', '1', None, None, 'unix (connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*"),',
'(connect, receive, send) type=stream peer=(label=unconfined,addr="@/tmp/dbus-*")', # noqa: E127
None)), # noqa: E127
('unixlike', False),
('deny unixlike,', False),
('priority=1 deny unixlike,', False),
)

View File

@ -174,6 +174,10 @@ class InvalidRlimitInit(AATest):
with self.assertRaises(AppArmorBug):
RlimitRule('as', '1024MB', audit=True)
def test_priority_keyword(self):
with self.assertRaises(AppArmorBug):
RlimitRule('as', '1024MB', priority=0)
class InvalidRlimitTest(AATest):
def _check_invalid_rawrule(self, rawrule):

View File

@ -277,6 +277,10 @@ class WriteSignalTestAATest(AATest):
('signal receive peer=foo,', 'signal receive peer=foo,'),
('signal (send receive) peer=/usr/bin/bar,', 'signal (receive send) peer=/usr/bin/bar,'),
('signal wr set=(pipe, usr1) peer=/sbin/baz,', 'signal wr set=(pipe usr1) peer=/sbin/baz,'),
('priority = 29 signal receive peer=foo,', 'priority=29 signal receive peer=foo,'),
('priority = 0 signal receive peer=foo,', 'priority=0 signal receive peer=foo,'),
('priority =-123 signal receive peer=foo,', 'priority=-123 signal receive peer=foo,'),
('priority =+10 signal receive peer=foo,', 'priority=10 signal receive peer=foo,'),
)
def test_write_manually(self):

View File

@ -166,15 +166,18 @@ class UnixTestGlob(AATest):
class UnixTestClean(AATest):
tests = (
(' audit unix , # foo ', 'audit unix, # foo'),
(' audit deny unix label = foo , ', 'audit deny unix label=foo,'),
(' audit allow unix peer = (addr = a) , # foo ', 'audit allow unix peer=(addr=a), # foo'),
(' deny unix type = foo , ', 'deny unix type=foo,'),
(' allow unix peer = (label=bb) , # foo ', 'allow unix peer=(label=bb), # foo'),
(' unix , # foo ', 'unix, # foo'),
(' unix addr = foo , ', 'unix addr=foo,'),
(' unix ( accept , rw) protocol = AA type = BB opt = myopt label = bb peer = (addr = a label = bb ) , ', 'unix (accept, rw) type=BB protocol=AA label=bb opt=myopt peer=(addr=a label=bb),'),
(' audit unix , # foo ', 'audit unix, # foo'),
(' audit deny unix label = foo , ', 'audit deny unix label=foo,'),
(' audit allow unix peer = (addr = a) , # foo ', 'audit allow unix peer=(addr=a), # foo'),
(' deny unix type = foo , ', 'deny unix type=foo,'),
(' allow unix peer = (label=bb) , # foo ', 'allow unix peer=(label=bb), # foo'),
(' unix , # foo ', 'unix, # foo'),
(' unix addr = foo , ', 'unix addr=foo,'),
(' unix ( accept , rw) protocol = AA type = BB opt = myopt label = bb peer = (addr = a label = bb ) , ', 'unix (accept, rw) type=BB protocol=AA label=bb opt=myopt peer=(addr=a label=bb),'),
('priority=-42 unix ( accept , rw) protocol = AA type = BB opt = myopt label = bb peer = (addr = a label = bb ), ', 'priority=-42 unix (accept, rw) type=BB protocol=AA label=bb opt=myopt peer=(addr=a label=bb),'),
('priority = 0 unix ( accept , rw) protocol = AA type = BB opt = myopt label = bb peer = (addr = a label = bb ), ', 'priority=0 unix (accept, rw) type=BB protocol=AA label=bb opt=myopt peer=(addr=a label=bb),'),
('priority=211 unix ( accept , rw) protocol = AA type = BB opt = myopt label = bb peer = (addr = a label = bb ), ', 'priority=211 unix (accept, rw) type=BB protocol=AA label=bb opt=myopt peer=(addr=a label=bb),'),
('priority=+45 unix ( accept , rw) protocol = AA type = BB opt = myopt label = bb peer = (addr = a label = bb ), ', 'priority=45 unix (accept, rw) type=BB protocol=AA label=bb opt=myopt peer=(addr=a label=bb),'),
)
def _run_test(self, rawrule, expected):

View File

@ -59,10 +59,10 @@ class UserNamespaceTestParseInvalid(AATest):
UserNamespaceRule.create_instance('foo,')
def test_diff_non_usernsrule(self):
exp = namedtuple('exp', ('audit', 'deny'))
exp = namedtuple('exp', ('audit', 'deny', 'priority'))
obj = UserNamespaceRule(('create'))
with self.assertRaises(AppArmorBug):
obj.is_equal(exp(False, False), False)
obj.is_equal(exp(False, False, None), False)
def test_diff_access(self):
obj1 = UserNamespaceRule(UserNamespaceRule.ALL)
@ -98,6 +98,10 @@ class WriteUserNamespaceTestAATest(AATest):
(' allow userns create ,# foo bar', 'allow userns create, # foo bar'),
('userns,', 'userns,'),
('userns create,', 'userns create,'),
(' priority = -1 allow userns create,', 'priority=-1 allow userns create,'),
(' priority = 0 allow userns create,', 'priority=0 allow userns create,'),
(' priority=+234 allow userns create,', 'priority=234 allow userns create,'),
(' priority = 65 allow userns create,', 'priority=65 allow userns create,'),
)
def _run_test(self, rawrule, expected):

View File

@ -30,10 +30,11 @@ exp = namedtuple('exp', ('comment', 'varname', 'mode', 'values'))
class VariableTest(AATest):
def _compare_obj(self, obj, expected):
# variables don't support the allow, audit or deny keyword
# variables don't support the allow, audit, deny, or priority keyword
self.assertEqual(False, obj.allow_keyword)
self.assertEqual(False, obj.audit)
self.assertEqual(False, obj.deny)
self.assertEqual(None, obj.priority)
self.assertEqual(expected.varname, obj.varname)
self.assertEqual(expected.mode, obj.mode)
@ -166,6 +167,14 @@ class InvalidVariableInit(AATest):
with self.assertRaises(AppArmorBug):
VariableRule('@{foo}', '=', '/bar', deny=True)
def test_invalid_priority_1(self):
with self.assertRaises(AppArmorBug):
VariableRule('@{foo}', '=', '/bar', priority=98)
def test_invalid_priority_2(self):
with self.assertRaises(AppArmorBug):
VariableRule('@{foo}', '=', '/bar', priority=0)
class InvalidVariableTest(AATest):
def _check_invalid_rawrule(self, rawrule, matches_regex=False):