Hi Christian, First off, let me say thanks for driving forward on this and not getting put off by my feedback.
Second is that I think these changes are large enough to not be
acceptable for 2.9.1, and that we should branch off 2.9.x before
committing this patch set.
> The updated patch (v4) is attached.
>
> Changes since v3:
> - document keys used in aa[profile][hat]
> - fix crash in write_capabilities()
> - improve resetting capability ruleset object in write_prior_segments()
> and serialize_profile_from_old_profile() (includes correct handling
> for include rules "for free")
>
>
> The usual line statistics:
> v4-1-add-base-and-capability-rule-class.diff - 371 lines added, 0 removed
> v4-2-add-capability-rule-test.diff - 806 lines added, 0 removed
> v4-3-use-capability-rule-class.diff - 68 lines added, 112 removed
Attached is a patch that applies on top of your patch series. Consider
it the cumulative critique of your patch set; in particular it tries to
address what I consider to be the most problematic bit of your patches,
namely that CapabilityRule objects only get partially initialized,
as I'm convinced it will lead to bugs. It requires that callers to
the class know whether or not it has been completely set up or not,
and leads to other funny issues like "what happens when set_param()
is called twice"? (I know what would happen, but I don't know if
that's the behavior you intend or if an exception should be raised;
there's no test case for that situation.)
As for converting log events into Rule objects, I'm of two minds. One
the one hand, this is exactly what is done after applying this
patch set, but the conversion occurs outside of the CapabilityRule
class. On the other, comparing two capability rules is easy; detecting
if a generic log event is covered is a simpler problem than detecting
whether one rule is completely covered by another, when both rules can
contain variables and regular expressions. Log events have an explicit
object, and thus an is_covered() test needs to only see if a given
string is matched by a regex, not whether two regexs cover the same
set of strings, or that one is a subset of the other. Thus, I've left
a bit of the code and the tests for log matching in but commented out.
A summary of the changes:
CapabilityRule/BaseRule
- complete initialization of CapabilityRule in __init__(); drop _init_vars()
- use standard pythonic notions of using a superclass' __init__ function.
- external user that want to specify the capability all rule should
pass CapabilityRule.ALL as the argument to the initializer.
- renamed inlinecomment to just comment
- converted get_raw() and get_clean() to use depth=0 by default for
convenience
- get rid of set_log and set_param as not needed
- move the implementation of set_raw to parser.py:parse_capability()
which returns a completed CapabilityRule object.
- rename parse_audit_allow to parse_modifiers and place in parser.py.
- add check to ensure the passed rule object is of CapabilityType for
is_covered and is_equal_localvars() (would prefer to be just
is_equal and call self.is_equal_modifiers within it).
CapabilityRuleset
- drop new_rule() as its not (currently) needed; every location
that wants a new rule has the context to know what type it is. If
that becomes not the case, having a ruleset class method that
points to the parser.py parse function for the rule type is the
approach I was thinking of taking, to cope with the differing
types of arguments that each rule's __init__ would need.
BaseRuleset
- drop add_raw, delete_raw, is_raw_covered, is_log_covered as not
necessary (see discussion above on is_log_covered)
- rename add_obj to add, delete_obj to delete, is_obj_covered to
is_covered, as interacting via Rule objects is the primary behavior.
- fixed up counting in delete() so argument to pop() wouldn't need to be
modified by -1.
aa.py
- convert to using new interfaces
- drop parse_audit_allow and use parser.py:parse_modifiers() instead.
I've also attached all four patches folded together (as
folded_capability_rules.patch), to let people see what the resulting
added files look like. The total diffstat looks like:
apparmor/aa.py | 190 67 123 0 +++------
apparmor/cleanprofile.py | 15 2 13 0
apparmor/parser.py | 76 76 0 0 +++
apparmor/rule/__init__.py | 189 189 0 0 +++++++++
apparmor/rule/capability.py | 119 119 0 0 ++++++
test/test-capability.py | 866 866 0 0
++++++++++++++++++++++++++++++++++++++++++++
6 files changed, 1319 insertions(+), 136 deletions(-)
--
Steve Beattie
<[email protected]>
http://NxNW.org/~steve/
utils: don't leave CapabilityRule half-initialized Signed-off-by: Steve Beattie <[email protected]> --- utils/apparmor/aa.py | 61 +---- utils/apparmor/parser.py | 76 +++++++ utils/apparmor/rule/__init__.py | 142 +++---------- utils/apparmor/rule/capability.py | 95 ++++---- utils/test/test-capability.py | 408 +++++++++++++++++++++----------------- 5 files changed, 416 insertions(+), 366 deletions(-) Index: b/utils/apparmor/rule/capability.py =================================================================== --- a/utils/apparmor/rule/capability.py +++ b/utils/apparmor/rule/capability.py @@ -1,3 +1,4 @@ +#!/usr/bin/env python # ---------------------------------------------------------------------- # Copyright (C) 2013 Kshitij Gupta <[email protected]> # Copyright (C) 2014 Christian Boltz <[email protected]> @@ -13,10 +14,8 @@ # # ---------------------------------------------------------------------- -from apparmor.regex import RE_PROFILE_CAP -from apparmor.common import AppArmorBug, AppArmorException +from apparmor.common import AppArmorBug from apparmor.rule import BaseRule, BaseRuleset -import re # setup module translations from apparmor.translations import init_translation @@ -26,61 +25,59 @@ _ = init_translation() class CapabilityRule(BaseRule): '''Class to handle and store a single capability rule''' - def _init_vars(self): - self.capability = set() # enforces in-rule deduplicates ("capability chown chown,") + # Nothing external should reference this class, all external users + # should reference the class field CapabilityRule.ALL + class __CapabilityAll(object): + pass + + ALL = __CapabilityAll + + def __init__(self, cap_list, audit=False, deny=False, allow_keyword=False, + comment='', log_event=None, raw_rule=None): + + super(CapabilityRule, self).__init__(audit=audit, deny=deny, + allow_keyword=allow_keyword, + comment=comment, + log_event=log_event, + raw_rule=raw_rule) + # Because we support having multiple caps in one rule, + # initializer needs to accept a list of caps. self.all_caps = False + if cap_list == CapabilityRule.ALL: + self.all_caps = True + self.capability = set() + else: + if type(cap_list) == str: + self.capability = {cap_list} + elif type(cap_list) == list and len(cap_list) > 0: + self.capability = set(cap_list) + else: + raise AppArmorBug('Passed unknown object to CapabilityRule: %s' % str(cap_list)) + # make sure none of the cap_list arguments are blank, in + # case we decide to return one cap per output line + for cap in self.capability: + if len(cap.strip()) == 0: + raise AppArmorBug('Passed empty capability to CapabilityRule: %s' % str(cap_list)) - def get_clean(self, depth): + def get_clean(self, depth=0): '''return rule (in clean/default formatting)''' space = ' ' * depth if self.all_caps: - return('%s%scapability,%s' % (space, self.modifiers_str(), self.inlinecomment)) + return('%s%scapability,%s' % (space, self.modifiers_str(), self.comment)) else: caps = ' '.join(self.capability).strip() # XXX return multiple lines, one for each capability, instead? if caps: - return('%s%scapability %s,%s' % (space, self.modifiers_str(), ' '.join(sorted(self.capability)), self.inlinecomment)) + return('%s%scapability %s,%s' % (space, self.modifiers_str(), ' '.join(sorted(self.capability)), self.comment)) else: raise AppArmorBug("Empty capability rule") - def set_raw(self, rawrule): - '''parse and store rawrule''' - - matches = RE_PROFILE_CAP.search(rawrule) - if not matches: - raise AppArmorException(_("Invalid capability rule '%s'") % rawrule) - - self.rawrule = rawrule.strip() - - self.audit, self.deny, self.allow_keyword, self.inlinecomment = self.parse_audit_allow(matches) - - if matches.group('capability'): - capability = matches.group('capability').strip() - self.capability = set(re.split("[ \t]+", capability)) - else: - self.all_caps = True - - def set_log(self, parsed_log_event): - '''parse and store log event''' - - if parsed_log_event['operation'] != 'capable': - raise AppArmorBug('Invalid capability log event - not a capability event') - - if not parsed_log_event['name']: - raise AppArmorBug('Invalid capability log event - name is empty') - - self.capability = {parsed_log_event['name']} - - self.logmode = parsed_log_event['aamode'] - - def set_param(self, capability, audit=False, deny=False): - self.capability = {capability} - self.audit = audit - self.deny = deny - def is_covered(self, rule_obj, check_allow_deny=True, check_audit=False): '''check if rule_obj is covered by this rule object''' + if not type(rule_obj) == CapabilityRule: + raise AppArmorBug('Passes non-capability rule: %s' % str(rule_obj)) + if check_allow_deny and self.deny != rule_obj.deny: return False @@ -105,8 +102,11 @@ class CapabilityRule(BaseRule): def is_equal_localvars(self, rule_obj): '''compare if rule-specific variables are equal''' - if ( self.capability != rule_obj.capability - or self.all_caps != rule_obj.all_caps ): + if not type(rule_obj) == CapabilityRule: + raise AppArmorBug('Passes non-capability rule: %s' % str(rule_obj)) + + if (self.capability != rule_obj.capability + or self.all_caps != rule_obj.all_caps): return False return True @@ -114,11 +114,6 @@ class CapabilityRule(BaseRule): class CapabilityRuleset(BaseRuleset): '''Class to handle and store a collection of capability rules''' - def new_rule(self): - '''tiny helper function that allows to keep several functions to parent class''' - return CapabilityRule() - def get_glob(self, path_or_rule): '''Return the next possible glob. For capability rules, that's always "capability," (all capabilities)''' return 'capability,' - Index: b/utils/apparmor/rule/__init__.py =================================================================== --- a/utils/apparmor/rule/__init__.py +++ b/utils/apparmor/rule/__init__.py @@ -23,28 +23,22 @@ _ = init_translation() class BaseRule(object): '''Base class to handle and store a single rule''' - def __init__(self): - '''initialize variables needed by all rule types - Do not override in child class unless really needed - override _init_vars() instead''' - self.audit = False - self.deny = False - self.allow_keyword = False + def __init__(self, audit=False, deny=False, allow_keyword=False, + comment='', log_event=None, raw_rule=''): + '''initialize variables needed by all rule types''' + self.audit = audit + self.deny = deny + self.allow_keyword = allow_keyword - self.inlinecomment = '' + self.comment = comment - self.rawrule = '' - self.logmode = '' # only set by set_log() - - self._init_vars() + self.raw_rule = raw_rule.strip() if raw_rule else None + self.log_event = log_event - def _init_vars(self): - '''called by __init__() - override in child class to initialize more variables''' - raise AppArmorBug('_init_vars() needs to be implemented in the child class') - - def get_raw(self, depth): + def get_raw(self, depth=0): '''return raw rule (with original formatting, and leading whitespace in the depth parameter)''' - if self.rawrule: - return '%s%s' % (' ' * depth, self.rawrule) + if self.raw_rule: + return '%s%s' % (' ' * depth, self.raw_rule) else: return self.get_clean(depth) @@ -57,8 +51,8 @@ class BaseRule(object): if strict and ( self.allow_keyword != rule_obj.allow_keyword - or self.inlinecomment != rule_obj.inlinecomment - or self.rawrule != rule_obj.rawrule + or self.comment != rule_obj.comment + or self.raw_rule != rule_obj.raw_rule ): return False @@ -81,33 +75,6 @@ class BaseRule(object): return '%s%s' % (auditstr, allowstr) - def parse_audit_allow(self, matches): - '''returns audit, deny, allow_keyword and comment from the matches object - - audit, deny and allow_keyword are True/False - - comment is the comment with a leading space''' - audit = False - if matches.group('audit'): - audit = True - - deny = False - allow_keyword = False - - allowstr = matches.group('allow') - if allowstr: - if allowstr.strip() == 'allow': - allow_keyword = True - elif allowstr.strip() == 'deny': - deny = True - else: - raise AppArmorBug("Invalid allow/deny keyword %s" % allowstr) - - comment = '' - if matches.group('comment'): - # include a space so that we don't need to add it everywhere when writing the rule - comment = ' %s' % matches.group('comment') - - return (audit, deny, allow_keyword, comment) - class BaseRuleset(object): '''Base class to handle and store a collection of rules''' @@ -126,18 +93,11 @@ class BaseRuleset(object): '''called by __init__() and delete_all_rules() - override in child class to initialize more variables''' pass - def add_raw(self, rawrule): - '''parse rawrule (from profile file) and store it in a structured way''' - - newrule = self.new_rule() - newrule.set_raw(rawrule) - self.rules.append(newrule) - - def add_obj(self, rule_obj): + def add(self, rule): '''add a rule object''' - self.rules.append(rule_obj) + self.rules.append(rule) - def get_raw(self, depth): + def get_raw(self, depth=0): '''return all raw rules (if possible/not modified in their original formatting). Returns an array of lines, with depth * leading whitespace''' @@ -150,7 +110,7 @@ class BaseRuleset(object): return data - def get_clean(self, depth): + def get_clean(self, depth=0): '''return all rules (in clean/default formatting) Returns an array of lines, with depth * leading whitespace''' @@ -178,64 +138,48 @@ class BaseRuleset(object): return cleandata - def is_obj_covered(self, rule_obj, check_allow_deny=True, check_audit=False): - '''return True if rule_obj is covered by existing rules, otherwise False''' + def is_covered(self, rule, check_allow_deny=True, check_audit=False): + '''return True if rule is covered by existing rules, otherwise False''' - for rule in self.rules: - if rule.is_covered(rule_obj, check_allow_deny, check_audit): + for r in self.rules: + if r.is_covered(rule, check_allow_deny, check_audit): return True return False - def is_log_covered(self, parsed_log_event, check_allow_deny=True, check_audit=False): - '''return True if parsed_log_event is covered by existing rules, otherwise False''' +# def is_log_covered(self, parsed_log_event, check_allow_deny=True, check_audit=False): +# '''return True if parsed_log_event is covered by existing rules, otherwise False''' +# +# rule_obj = self.new_rule() +# rule_obj.set_log(parsed_log_event) +# +# return self.is_covered(rule_obj, check_allow_deny, check_audit) - rule_obj = self.new_rule() - rule_obj.set_log(parsed_log_event) + def delete(self, rule): + '''Delete rule from rules''' - return self.is_obj_covered(rule_obj, check_allow_deny, check_audit) - - def is_raw_covered(self, rawrule, check_allow_deny=True, check_audit=False): - '''return True if rawrule is covered by existing rules, otherwise False''' - - rule_obj = self.new_rule() - rule_obj.set_raw(rawrule) - - return self.is_obj_covered(rule_obj, check_allow_deny, check_audit) - - def delete_obj(self, rule_obj): - '''Delete rule_obj from rules''' - - rule_to_delete = None + rule_to_delete = False i = 0 - for rule in self.rules: - i = i + 1 - if rule.is_equal(rule_obj): - rule_to_delete = i + for r in self.rules: + if r.is_equal(rule): + rule_to_delete = True break + i = i + 1 if rule_to_delete: - self.rules.pop(i-1) + self.rules.pop(i) else: - raise AppArmorBug('Attemp to delete non-existing rule %s' % rule_obj.get_raw(0)) - - def delete_raw(self, rawrule): - '''Delete rawrule from rules''' - - rule_obj = self.new_rule() - rule_obj.set_raw(rawrule) - - return self.delete_obj(rule_obj) + raise AppArmorBug('Attempt to delete non-existing rule %s' % rule.get_raw(0)) def delete_duplicates(self, include_rules): '''Delete duplicate rules. include_rules must be a *_rules object''' deleted = [] if include_rules: # avoid breakage until we have a proper function to ensure all profiles contain all *_rules objects - for rule_obj in self.rules: - if include_rules.is_obj_covered(rule_obj, True, True): - self.delete_obj(rule_obj) - deleted.append(rule_obj) + for rule in self.rules: + if include_rules.is_covered(rule, True, True): + self.delete(rule) + deleted.append(rule) return len(deleted) @@ -243,5 +187,3 @@ class BaseRuleset(object): '''returns the next possible glob with extension (for file rules only). For all other rule types, raise an exception''' raise AppArmorBug("get_glob_ext is not available for this rule type!") - - Index: b/utils/test/test-capability.py =================================================================== --- a/utils/test/test-capability.py +++ b/utils/test/test-capability.py @@ -16,6 +16,8 @@ import unittest from apparmor.rule.capability import CapabilityRule, CapabilityRuleset +from apparmor.parser import parse_capability, parse_modifiers +from apparmor.rule import BaseRule from apparmor.common import AppArmorException, AppArmorBug, hasher from apparmor.logparser import ReadLog @@ -29,10 +31,9 @@ class CapabilityTest(unittest.TestCase): def _compare_obj_with_rawrule(self, rawrule, expected): - obj = CapabilityRule() - obj.set_raw(rawrule) + obj = parse_capability(rawrule) - self.assertEqual(rawrule.strip(), obj.rawrule) + self.assertEqual(rawrule.strip(), obj.raw_rule) self._compare_obj(obj, expected) @@ -42,7 +43,7 @@ class CapabilityTest(unittest.TestCase): self.assertEqual(expected['capability'], obj.capability) self.assertEqual(expected['all_caps'], obj.all_caps) self.assertEqual(expected['deny'], obj.deny) - self.assertEqual(expected['inlinecomment'], obj.inlinecomment) + self.assertEqual(expected['inlinecomment'], obj.comment) def test_cap_allow_all(self): self._compare_obj_with_rawrule("capability,", { @@ -120,8 +121,7 @@ class CapabilityTest(unittest.TestCase): 'name': 'net_raw' }) - obj = CapabilityRule() - obj.set_log(parsed_event) + obj = CapabilityRule(parsed_event['name'], log_event=parsed_event) self._compare_obj(obj, { 'allow_keyword': False, @@ -134,39 +134,50 @@ class CapabilityTest(unittest.TestCase): self.assertEqual(obj.get_raw(1), ' capability net_raw,') - def test_cap_from_invalid_log(self): - parser = ReadLog('', '', '', '', '') - # invalid log entry, name= should contain the capability name - event = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname=""' - - parsed_event = parser.parse_event(event) +# def test_cap_from_invalid_log(self): +# parser = ReadLog('', '', '', '', '') +# # invalid log entry, name= should contain the capability name +# event = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname=""' +# +# parsed_event = parser.parse_event(event) +# +# obj = CapabilityRule() +# +# with self.assertRaises(AppArmorBug): +# obj.set_log(parsed_event) +# +# with self.assertRaises(AppArmorBug): +# obj.get_raw(1) +# +# def test_cap_from_non_cap_log(self): +# parser = ReadLog('', '', '', '', '') +# # log entry for different rule type +# event = 'type=AVC msg=audit(1415403814.973:667): apparmor="ALLOWED" operation="setsockopt" profile="/home/sys-tmp/ping" pid=15454 comm="ping" lport=1 family="inet" sock_type="raw" protocol=1' +# +# parsed_event = parser.parse_event(event) +# +# obj = CapabilityRule() +# +# with self.assertRaises(AppArmorBug): +# obj.set_log(parsed_event) +# +# with self.assertRaises(AppArmorBug): +# obj.get_raw(1) - obj = CapabilityRule() - - with self.assertRaises(AppArmorBug): - obj.set_log(parsed_event) - - with self.assertRaises(AppArmorBug): - obj.get_raw(1) - - def test_cap_from_non_cap_log(self): - parser = ReadLog('', '', '', '', '') - # log entry for different rule type - event = 'type=AVC msg=audit(1415403814.973:667): apparmor="ALLOWED" operation="setsockopt" profile="/home/sys-tmp/ping" pid=15454 comm="ping" lport=1 family="inet" sock_type="raw" protocol=1' + def test_cap_from_init_01(self): + obj = CapabilityRule('chown') - parsed_event = parser.parse_event(event) - - obj = CapabilityRule() - - with self.assertRaises(AppArmorBug): - obj.set_log(parsed_event) - - with self.assertRaises(AppArmorBug): - obj.get_raw(1) + self._compare_obj(obj, { + 'allow_keyword': False, + 'deny': False, + 'audit': False, + 'capability': {'chown'}, + 'all_caps': False, + 'inlinecomment': "", + }) - def test_cap_from_param(self): - obj = CapabilityRule() - obj.set_param('chown') + def test_cap_from_init_02(self): + obj = CapabilityRule(['chown']) self._compare_obj(obj, { 'allow_keyword': False, @@ -177,9 +188,8 @@ class CapabilityTest(unittest.TestCase): 'inlinecomment': "", }) - def test_cap_from_param_2(self): - obj = CapabilityRule() - obj.set_param('chown', audit=True, deny=True) + def test_cap_from_init_03(self): + obj = CapabilityRule('chown', audit=True, deny=True) self._compare_obj(obj, { 'allow_keyword': False, @@ -190,15 +200,31 @@ class CapabilityTest(unittest.TestCase): 'inlinecomment': "", }) + def test_cap_from_init_04(self): + obj = CapabilityRule(['chown', 'fsetid'], deny=True) + + self._compare_obj(obj, { + 'allow_keyword': False, + 'deny': True, + 'audit': False, + 'capability': {'chown', 'fsetid'}, + 'all_caps': False, + 'inlinecomment': "", + }) + class InvalidCapabilityTest(unittest.TestCase): def setUp(self): self.maxDiff = None + # XXX - these tests more properly belong in a test suite for + # XXX - parser.py::parse_capability() def _check_invalid_rawrule(self, rawrule): - obj = CapabilityRule() + obj = None with self.assertRaises(AppArmorException): - obj.set_raw(rawrule) + obj = CapabilityRule(parse_capability(rawrule)) + + self.assertIsNone(obj, 'CapbilityRule handed back an object unexpectedly') def test_invalid_cap_missing_comma(self): self._check_invalid_rawrule('capability') # missing comma @@ -206,26 +232,45 @@ class InvalidCapabilityTest(unittest.Tes def test_invalid_cap_non_CapabilityRule(self): self._check_invalid_rawrule('network,') # not a capability rule - def test_empty_cap_list(self): - obj = CapabilityRule() + # XXX - this test more properly belongs in a test suite for + # XXX - parser.py::parse_modifiers() + def test_parse_modifiers_invalid(self): + regex = re.compile('^\s*(?P<audit>audit\s+)?(?P<allow>allow\s+|deny\s+|invalid\s+)?') + matches = regex.search('audit invalid ') + + with self.assertRaises(AppArmorBug): + parse_modifiers(matches) + + def test_empty_cap_set(self): + obj = CapabilityRule('chown') + obj.capability.clear() # no capability set, and ALL not set with self.assertRaises(AppArmorBug): obj.get_clean(1) + def test_empty_cap_list(self): + with self.assertRaises(AppArmorBug): + CapabilityRule([]) + + def test_no_cap_list_arg(self): + with self.assertRaises(TypeError): + CapabilityRule() + def test_space_cap(self): - obj = CapabilityRule() - obj.set_param(' ') # the whitespace capability ;-) with self.assertRaises(AppArmorBug): - obj.get_clean(1) + CapabilityRule(' ') # the whitespace capability ;-) - def test_parse_audit_allow_invalid(self): - obj = CapabilityRule() + def test_space_list_1(self): + with self.assertRaises(AppArmorBug): + CapabilityRule([' ', ' ', ' ']) # the whitespace capability ;-) - regex = re.compile('^\s*(?P<audit>audit\s+)?(?P<allow>allow\s+|deny\s+|invalid\s+)?') - matches = regex.search('audit invalid ') + def test_space_list_2(self): + with self.assertRaises(AppArmorBug): + CapabilityRule(['chown', ' ', 'setgid']) # includes the whitespace capability ;-) + def test_wrong_type_for_cap(self): with self.assertRaises(AppArmorBug): - obj.parse_audit_allow(matches) + CapabilityRule(dict()) class WriteCapabilityTest(unittest.TestCase): @@ -233,10 +278,9 @@ class WriteCapabilityTest(unittest.TestC self.maxDiff = None def _check_write_rule(self, rawrule, cleanrule): - obj = CapabilityRule() - obj.set_raw(rawrule) - clean = obj.get_clean(0) - raw = obj.get_raw(0) + obj = parse_capability(rawrule) + clean = obj.get_clean() + raw = obj.get_raw() self.assertEqual(cleanrule.strip(), clean, 'unexpected clean rule') self.assertEqual(rawrule.strip(), raw, 'unexpected raw rule') @@ -251,9 +295,7 @@ class WriteCapabilityTest(unittest.TestC self._check_write_rule(' deny capability sys_admin audit_write,# foo bar', 'deny capability audit_write sys_admin, # foo bar') def test_write_manually(self): - obj = CapabilityRule() - obj.capability = {'ptrace', 'audit_write'} - obj.allow_keyword = True + obj = CapabilityRule(['ptrace', 'audit_write'], allow_keyword=True) expected = ' allow capability audit_write ptrace,' @@ -264,22 +306,17 @@ class CapabilityCoveredTest(unittest.Tes def setUp(self): self.maxDiff = None - def _init_obj_with_rawrule(self, rawrule): - obj = CapabilityRule() - obj.set_raw(rawrule) - return obj - def _is_covered(self, obj, rule_to_test): - return obj.is_covered(self._init_obj_with_rawrule(rule_to_test)) + return obj.is_covered(parse_capability(rule_to_test)) def _is_covered_exact(self, obj, rule_to_test): - return obj.is_covered(self._init_obj_with_rawrule(rule_to_test), True, True) + return obj.is_covered(parse_capability(rule_to_test), True, True) def _is_equal(self, obj, rule_to_test, strict): - return obj.is_equal(self._init_obj_with_rawrule(rule_to_test), strict) + return obj.is_equal(parse_capability(rule_to_test), strict) def test_covered_single(self): - obj = self._init_obj_with_rawrule('capability sys_admin,') + obj = parse_capability('capability sys_admin,') self.assertTrue(self._is_covered(obj, 'capability sys_admin,')) @@ -289,7 +326,7 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_covered(obj, 'capability,')) def test_covered_audit(self): - obj = self._init_obj_with_rawrule('audit capability sys_admin,') + obj = parse_capability('audit capability sys_admin,') self.assertTrue(self._is_covered(obj, 'capability sys_admin,')) self.assertTrue(self._is_covered(obj, 'audit capability sys_admin,')) @@ -299,7 +336,7 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_covered(obj, 'capability,')) def test_covered_check_audit(self): - obj = self._init_obj_with_rawrule('audit capability sys_admin,') + obj = parse_capability('audit capability sys_admin,') self.assertFalse(self._is_covered_exact(obj, 'capability sys_admin,')) self.assertTrue(self._is_covered_exact(obj, 'audit capability sys_admin,')) @@ -309,7 +346,7 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_covered_exact(obj, 'capability,')) def test_equal(self): - obj = self._init_obj_with_rawrule('capability sys_admin,') + obj = parse_capability('capability sys_admin,') self.assertTrue(self._is_equal(obj, 'capability sys_admin,', True)) self.assertFalse(self._is_equal(obj, 'allow capability sys_admin,', True)) @@ -321,7 +358,7 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_equal(obj, 'audit capability sys_admin,', False)) def test_covered_multi(self): - obj = self._init_obj_with_rawrule('capability audit_write sys_admin,') + obj = parse_capability('capability audit_write sys_admin,') self.assertTrue(self._is_covered(obj, 'capability sys_admin,')) self.assertTrue(self._is_covered(obj, 'capability audit_write,')) @@ -333,7 +370,7 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_covered(obj, 'capability,')) def test_covered_all(self): - obj = self._init_obj_with_rawrule('capability,') + obj = parse_capability('capability,') self.assertTrue(self._is_covered(obj, 'capability sys_admin,')) self.assertTrue(self._is_covered(obj, 'capability audit_write,')) @@ -344,7 +381,7 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_covered(obj, 'audit capability,')) def test_covered_deny(self): - obj = self._init_obj_with_rawrule('capability sys_admin,') + obj = parse_capability('capability sys_admin,') self.assertTrue(self._is_covered(obj, 'capability sys_admin,')) @@ -354,7 +391,7 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_covered(obj, 'capability,')) def test_covered_deny_2(self): - obj = self._init_obj_with_rawrule('deny capability sys_admin,') + obj = parse_capability('deny capability sys_admin,') self.assertTrue(self._is_covered(obj, 'deny capability sys_admin,')) @@ -363,22 +400,40 @@ class CapabilityCoveredTest(unittest.Tes self.assertFalse(self._is_covered(obj, 'deny capability chown,')) self.assertFalse(self._is_covered(obj, 'deny capability,')) - def test_invalid_is_obj_covered(self): - obj = self._init_obj_with_rawrule('capability sys_admin,') + def test_invalid_is_covered(self): + obj = parse_capability('capability sys_admin,') - testobj = CapabilityRule() # no capability set + testobj = BaseRule() # different type with self.assertRaises(AppArmorBug): obj.is_covered(testobj) + def test_borked_obj_is_covered(self): + obj = parse_capability('capability sys_admin,') + + testobj = CapabilityRule('chown') + testobj.capability.clear() + + with self.assertRaises(AppArmorBug): + obj.is_covered(testobj) + + def test_invalid_is_equal(self): + obj = parse_capability('capability sys_admin,') + + testobj = BaseRule() # different type + + with self.assertRaises(AppArmorBug): + obj.is_equal(testobj) + def test_empty_init(self): # add to internal set instead of using .set_* (which overwrites the internal set) to make sure obj and obj2 use separate storage - obj = CapabilityRule() + obj = CapabilityRule('fsetid') + obj2 = CapabilityRule('fsetid') obj.capability.add('sys_admin') - - obj2 = CapabilityRule() obj2.capability.add('ptrace') + self.assertTrue(self._is_covered(obj, 'capability sys_admin,')) + self.assertFalse(self._is_covered(obj, 'capability ptrace,')) self.assertFalse(self._is_covered(obj2, 'capability sys_admin,')) self.assertTrue(self._is_covered(obj2, 'capability ptrace,')) @@ -416,10 +471,10 @@ class CapabilityRulesTest(unittest.TestC ] for rule in rules: - ruleset.add_raw(rule) + ruleset.add(parse_capability(rule)) - self.assertEqual(expected_raw, ruleset.get_raw(0)) - self.assertEqual(expected_clean, ruleset.get_clean(0)) + self.assertEqual(expected_raw, ruleset.get_raw()) + self.assertEqual(expected_clean, ruleset.get_clean()) def test_ruleset_2(self): ruleset = CapabilityRuleset() @@ -445,17 +500,16 @@ class CapabilityRulesTest(unittest.TestC ] for rule in rules: - ruleset.add_raw(rule) + ruleset.add(parse_capability(rule)) self.assertEqual(expected_raw, ruleset.get_raw(1)) self.assertEqual(expected_clean, ruleset.get_clean(1)) - def test_ruleset_add_obj(self): - rule_obj = CapabilityRule() - rule_obj.set_raw('capability chgrp, # example comment') + def test_ruleset_add(self): + rule = CapabilityRule('chgrp', comment=' # example comment') ruleset = CapabilityRuleset() - ruleset.add_obj(rule_obj) + ruleset.add(rule) expected_raw = [ ' capability chgrp, # example comment', @@ -482,75 +536,81 @@ class CapabilityRulesCoveredTest(unittes ] for rule in rules: - self.ruleset.add_raw(rule) - - def test_ruleset_is_raw_covered_1(self): - self.assertTrue(self.ruleset.is_raw_covered('capability chown,')) - def test_ruleset_is_raw_covered_2(self): - self.assertTrue(self.ruleset.is_raw_covered('capability sys_admin,')) - def test_ruleset_is_raw_covered_3(self): - self.assertTrue(self.ruleset.is_raw_covered('allow capability sys_admin,')) - def test_ruleset_is_raw_covered_4(self): - self.assertTrue(self.ruleset.is_raw_covered('capability setuid,')) - def test_ruleset_is_raw_covered_5(self): - self.assertTrue(self.ruleset.is_raw_covered('allow capability setgid,')) - def test_ruleset_is_raw_covered_6(self): - self.assertTrue(self.ruleset.is_raw_covered('capability setgid setuid,')) - def test_ruleset_is_raw_covered_7(self): - pass # self.assertTrue(self.ruleset.is_raw_covered('capability sys_admin chown,')) # fails because it is split over two rule objects internally - def test_ruleset_is_raw_covered_8(self): - self.assertTrue(self.ruleset.is_raw_covered('capability kill,')) - - def test_ruleset_is_raw_covered_9(self): - self.assertFalse(self.ruleset.is_raw_covered('deny capability chown,')) - def test_ruleset_is_raw_covered_10(self): - self.assertFalse(self.ruleset.is_raw_covered('deny capability sys_admin,')) - def test_ruleset_is_raw_covered_11(self): - self.assertFalse(self.ruleset.is_raw_covered('deny capability sys_admin chown,')) - def test_ruleset_is_raw_covered_12(self): - self.assertFalse(self.ruleset.is_raw_covered('deny capability setgid,')) - def test_ruleset_is_raw_covered_13(self): - self.assertFalse(self.ruleset.is_raw_covered('deny capability kill,')) - - def test_ruleset_is_raw_covered_14(self): - self.assertFalse(self.ruleset.is_raw_covered('audit capability chown,')) - def test_ruleset_is_raw_covered_15(self): - self.assertFalse(self.ruleset.is_raw_covered('audit capability sys_admin,')) - def test_ruleset_is_raw_covered_16(self): - self.assertFalse(self.ruleset.is_raw_covered('audit capability sys_admin chown,')) - def test_ruleset_is_raw_covered_17(self): - self.assertFalse(self.ruleset.is_raw_covered('audit capability setgid,')) - def test_ruleset_is_raw_covered_18(self): - self.assertTrue(self.ruleset.is_raw_covered('audit capability kill,')) - - def test_ruleset_is_raw_covered_19(self): - self.assertTrue(self.ruleset.is_raw_covered('deny capability chgrp,')) - def test_ruleset_is_raw_covered_20(self): - self.assertFalse(self.ruleset.is_raw_covered('audit deny capability chgrp,')) - def test_ruleset_is_raw_covered_21(self): - self.assertFalse(self.ruleset.is_raw_covered('audit capability chgrp,')) - - def _test_log_covered(self, expected, capability): - event_base = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname="%s"' + self.ruleset.add(parse_capability(rule)) - parser = ReadLog('', '', '', '', '') - self.assertEqual(expected, self.ruleset.is_log_covered(parser.parse_event(event_base%capability))) - - def test_ruleset_is_log_covered_1(self): - self._test_log_covered(False, 'net_raw') - def test_ruleset_is_log_covered_2(self): - self._test_log_covered(True, 'chown') - def test_ruleset_is_log_covered_3(self): - self._test_log_covered(True, 'sys_admin') - def test_ruleset_is_log_covered_4(self): - self._test_log_covered(True, 'kill') - def test_ruleset_is_log_covered_5(self): - self._test_log_covered(False, 'chgrp') - def test_ruleset_is_log_covered_6(self): - event_base = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname="%s"' - - parser = ReadLog('', '', '', '', '') - self.assertEqual(True, self.ruleset.is_log_covered(parser.parse_event(event_base%'chgrp'), False)) # ignores allow/deny + def test_ruleset_is_covered_1(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('capability chown,'))) + def test_ruleset_is_covered_2(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('capability sys_admin,'))) + def test_ruleset_is_covered_3(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('allow capability sys_admin,'))) + def test_ruleset_is_covered_4(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('capability setuid,'))) + def test_ruleset_is_covered_5(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('allow capability setgid,'))) + def test_ruleset_is_covered_6(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('capability setgid setuid,'))) + def test_ruleset_is_covered_7(self): + pass # self.assertTrue(self.ruleset.is_covered(parse_capability('capability sys_admin chown,'))) # fails because it is split over two rule objects internally + def test_ruleset_is_covered_8(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('capability kill,'))) + + def test_ruleset_is_covered_9(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability chown,'))) + def test_ruleset_is_covered_10(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability sys_admin,'))) + def test_ruleset_is_covered_11(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability sys_admin chown,'))) + def test_ruleset_is_covered_12(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability setgid,'))) + def test_ruleset_is_covered_13(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability kill,'))) + + def test_ruleset_is_covered_14(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability chown,'))) + def test_ruleset_is_covered_15(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability sys_admin,'))) + def test_ruleset_is_covered_16(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability sys_admin chown,'))) + def test_ruleset_is_covered_17(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability setgid,'))) + def test_ruleset_is_covered_18(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('audit capability kill,'))) + + def test_ruleset_is_covered_19(self): + self.assertTrue(self.ruleset.is_covered(parse_capability('deny capability chgrp,'))) + def test_ruleset_is_covered_20(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('audit deny capability chgrp,'))) + def test_ruleset_is_covered_21(self): + self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability chgrp,'))) + +# XXX - disabling these until we decide whether or not checking whether +# a log is covered by rules should be a separate entry point, possibly +# handling the log structure directly, or whether coverage should be +# solely based on Rule objects and marshaling of a log message into a +# Rule object should occur outside of the Rule classes themselves. +# +# def _test_log_covered(self, expected, capability): +# event_base = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname="%s"' + +# parser = ReadLog('', '', '', '', '') +# self.assertEqual(expected, self.ruleset.is_log_covered(parser.parse_event(event_base%capability))) +# +# def test_ruleset_is_log_covered_1(self): +# self._test_log_covered(False, 'net_raw') +# def test_ruleset_is_log_covered_2(self): +# self._test_log_covered(True, 'chown') +# def test_ruleset_is_log_covered_3(self): +# self._test_log_covered(True, 'sys_admin') +# def test_ruleset_is_log_covered_4(self): +# self._test_log_covered(True, 'kill') +# def test_ruleset_is_log_covered_5(self): +# self._test_log_covered(False, 'chgrp') +# def test_ruleset_is_log_covered_6(self): +# event_base = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname="%s"' +# +# parser = ReadLog('', '', '', '', '') +# self.assertEqual(True, self.ruleset.is_log_covered(parser.parse_event(event_base%'chgrp'), False)) # ignores allow/deny class CapabilityGlobTest(unittest.TestCase): def setUp(self): @@ -576,9 +636,9 @@ class CapabilityDeleteTest(unittest.Test ] for rule in rules: - self.ruleset.add_raw(rule) + self.ruleset.add(parse_capability(rule)) - def test_delete_raw(self): + def test_delete(self): expected_raw = [ ' capability chown,', ' deny capability chgrp, # example comment', @@ -592,12 +652,12 @@ class CapabilityDeleteTest(unittest.Test '', ] - self.ruleset.delete_raw('capability sys_admin,') + self.ruleset.delete(CapabilityRule(['sys_admin'])) self.assertEqual(expected_raw, self.ruleset.get_raw(1)) self.assertEqual(expected_clean, self.ruleset.get_clean(1)) - def test_delete_raw_with_allcaps(self): + def test_delete_with_allcaps(self): expected_raw = [ ' capability chown,', ' deny capability chgrp, # example comment', @@ -613,13 +673,13 @@ class CapabilityDeleteTest(unittest.Test '', ] - self.ruleset.add_raw('capability,') - self.ruleset.delete_raw('capability sys_admin,') + self.ruleset.add(CapabilityRule(CapabilityRule.ALL)) + self.ruleset.delete(CapabilityRule('sys_admin')) self.assertEqual(expected_raw, self.ruleset.get_raw(1)) self.assertEqual(expected_clean, self.ruleset.get_clean(1)) - def test_delete_raw_with_multi(self): + def test_delete_with_multi(self): expected_raw = [ ' capability chown,', ' allow capability sys_admin,', @@ -635,23 +695,23 @@ class CapabilityDeleteTest(unittest.Test '', ] - self.ruleset.add_raw('capability audit_read audit_write,') - self.ruleset.delete_raw('capability audit_read audit_write,') + self.ruleset.add(CapabilityRule(['audit_read', 'audit_write'])) + self.ruleset.delete(CapabilityRule(['audit_read', 'audit_write'])) self.assertEqual(expected_raw, self.ruleset.get_raw(1)) self.assertEqual(expected_clean, self.ruleset.get_clean(1)) - def test_delete_raw_with_multi_2(self): - self.ruleset.add_raw('capability audit_read audit_write,') + def test_delete_with_multi_2(self): + self.ruleset.add(CapabilityRule(['audit_read', 'audit_write'])) with self.assertRaises(AppArmorBug): # XXX ideally delete_raw should remove audit_read from the "capability audit_read audit_write," ruleset # but that's quite some work to cover a corner case. - self.ruleset.delete_raw('capability audit_read,') + self.ruleset.delete(CapabilityRule('audit_read')) def test_delete_raw_notfound(self): with self.assertRaises(AppArmorBug): - self.ruleset.delete_raw('capability audit_write,') + self.ruleset.delete(CapabilityRule('audit_write')) def test_delete_duplicates(self): inc = CapabilityRuleset() @@ -661,7 +721,7 @@ class CapabilityDeleteTest(unittest.Test ] for rule in rules: - inc.add_raw(rule) + inc.add(parse_capability(rule)) expected_raw = [ ' allow capability sys_admin,', @@ -682,7 +742,7 @@ class CapabilityDeleteTest(unittest.Test ] for rule in rules: - inc.add_raw(rule) + inc.add(parse_capability(rule)) expected_raw = [ ' capability chown,', @@ -704,7 +764,7 @@ class CapabilityDeleteTest(unittest.Test self.assertEqual(expected_clean, self.ruleset.get_clean(1)) def test_delete_duplicates_3(self): - self.ruleset.add_raw('audit capability dac_override,') + self.ruleset.add(parse_capability('audit capability dac_override,')) inc = CapabilityRuleset() rules = [ @@ -712,7 +772,7 @@ class CapabilityDeleteTest(unittest.Test ] for rule in rules: - inc.add_raw(rule) + inc.add(parse_capability(rule)) expected_raw = [ ' capability chown,', @@ -742,7 +802,7 @@ class CapabilityDeleteTest(unittest.Test ] for rule in rules: - inc.add_raw(rule) + inc.add(parse_capability(rule)) expected_raw = [ ' allow capability sys_admin,', # XXX huh? should be deleted! Index: b/utils/apparmor/aa.py =================================================================== --- a/utils/apparmor/aa.py +++ b/utils/apparmor/aa.py @@ -53,6 +53,7 @@ from apparmor.regex import (RE_PROFILE_S import apparmor.rules as aarules from apparmor.rule.capability import CapabilityRuleset, CapabilityRule +import apparmor.parser as aaparser from apparmor.yasti import SendDataToYast, GetDataFromYast, shutdown_yast @@ -1554,8 +1555,7 @@ def ask_the_questions(): for hat in hats: for capability in sorted(log_dict[aamode][profile][hat]['capability'].keys()): # skip if capability already in profile - capability_obj = CapabilityRule() - capability_obj.set_param(capability) + capability_obj = CapabilityRule(capability) if is_known_rule(aa[profile][hat], 'capability', capability_obj): continue # Load variables? Don't think so. @@ -1636,9 +1636,8 @@ def ask_the_questions(): aaui.UI_Info(_('Deleted %s previous matching profile entries.') % deleted) else: - capability_obj = CapabilityRule() - capability_obj.set_param(capability, audit=audit) - aa[profile][hat]['capability'].add_obj(capability_obj) + capability_obj = CapabilityRule(capability, audit=audit) + aa[profile][hat]['capability'].add(capability_obj) aaui.UI_Info(_('Adding capability %s to profile.') % capability) changed[profile] = True @@ -1646,9 +1645,8 @@ def ask_the_questions(): done = True elif ans == 'CMD_DENY': - capability_obj = CapabilityRule() - capability_obj.set_param(capability, audit=audit, deny=True) - aa[profile][hat]['capability'].add_obj(capability_obj) + capability_obj = CapabilityRule(capability, audit=audit, deny=True) + aa[profile][hat]['capability'].add(capability_obj) changed[profile] = True aaui.UI_Info(_('Denying capability %s to profile.') % capability) @@ -2197,18 +2195,16 @@ def match_net_include(incname, family, t return False -def match_cap_includes(profile, cap): +def match_cap_includes(profile, capability): # still used by aa-mergeprof - capability_obj = CapabilityRule() - capability_obj.set_param(cap) + capability_obj = CapabilityRule(capability) return match_includes(profile, 'capability', capability_obj) def match_includes(profile, rule_type, rule_obj): newincludes = [] for incname in include.keys(): # XXX type check should go away once we init all profiles correctly - # XXX using a CapabilityRule object instead of is_raw_covered() might be nice/faster - if valid_include(profile, incname) and include[incname][incname].get(rule_type, False) and include[incname][incname][rule_type].is_obj_covered(rule_obj): + if valid_include(profile, incname) and include[incname][incname].get(rule_type, False) and include[incname][incname][rule_type].is_covered(rule_obj): newincludes.append(incname) return newincludes @@ -2516,11 +2512,11 @@ def collapse_log(): log_dict[aamode][profile][hat]['path'][path] = mode - for capability in prelog[aamode][profile][hat]['capability'].keys(): + for cap in prelog[aamode][profile][hat]['capability'].keys(): # If capability not already in profile # XXX remove first check when we have proper profile initialisation - if aa[profile][hat].get('capability', False) and not aa[profile][hat]['capability'].is_raw_covered("capability %s," % capability): - log_dict[aamode][profile][hat]['capability'][capability] = True + if aa[profile][hat].get('capability', False) and not aa[profile][hat]['capability'].is_covered(CapabilityRule(cap)): + log_dict[aamode][profile][hat]['capability'][cap] = True nd = prelog[aamode][profile][hat]['netdomain'] for family in nd.keys(): @@ -2733,7 +2729,7 @@ def parse_profile_data(data, file, do_in if not profile_data[profile][hat].get('capability', False): profile_data[profile][hat]['capability'] = CapabilityRuleset() - profile_data[profile][hat]['capability'].add_raw(line) + profile_data[profile][hat]['capability'].add(aaparser.parse_capability(line)) elif RE_PROFILE_LINK.search(line): matches = RE_PROFILE_LINK.search(line).groups() @@ -2842,7 +2838,7 @@ def parse_profile_data(data, file, do_in if not profile: raise AppArmorException(_('Syntax Error: Unexpected bare file rule found in file: %(file)s line: %(line)s') % { 'file': file, 'line': lineno + 1 }) - audit, allow, allow_keyword, comment = parse_audit_allow(matches) + audit, allow, allow_keyword, comment = aaparser.parse_modifiers(matches) # TODO: honor allow_keyword and comment mode = apparmor.aamode.AA_BARE_FILE_MODE @@ -3181,26 +3177,6 @@ def parse_profile_data(data, file, do_in return profile_data -def parse_audit_allow(matches): - audit = False - if matches.group('audit'): - audit = True - - allow = 'allow' - allow_keyword = False - if matches.group('allow'): - allow = matches.group('allow').strip() - allow_keyword = True - if allow != 'allow' and allow != 'deny': # should never happen - raise AppArmorException(_("Invalid allow/deny keyword %s" % allow)) - - comment = '' - if matches.group('comment'): - # include a space so that we don't need to add it everywhere when writing the rule - comment = ' %s' % matches.group('comment') - - return (audit, allow, allow_keyword, comment) - # RE_DBUS_ENTRY = re.compile('^dbus\s*()?,\s*$') # use stuff like '(?P<action>(send|write|w|receive|read|r|rw))' @@ -3929,11 +3905,12 @@ def serialize_profile_from_old_profile(p profile = None elif RE_PROFILE_CAP.search(line): - if write_prof_data[hat]['capability'].is_raw_covered(line, True, True): + cap = aaparser.parse_capability(line) + if write_prof_data[hat]['capability'].is_covered(cap, True, True): if not segments['capability'] and True in segments.values(): data += write_prior_segments(write_prof_data[name], segments, line) segments['capability'] = True - write_prof_data[hat]['capability'].delete_raw(line) + write_prof_data[hat]['capability'].delete(cap) data.append(line) else: # To-Do @@ -4313,12 +4290,12 @@ def profile_known_exec(profile, typ, exe def is_known_rule(profile, rule_type, rule_obj): # XXX get rid of get() checks after we have a proper function to initialize a profile if profile.get(rule_type, False): - if profile[rule_type].is_obj_covered(rule_obj, False): + if profile[rule_type].is_covered(rule_obj, False): return True for incname in profile['include'].keys(): if include[incname][incname].get(rule_type, False): - if include[incname][incname][rule_type].is_obj_covered(rule_obj, False): + if include[incname][incname][rule_type].is_covered(rule_obj, False): return True return False Index: b/utils/apparmor/parser.py =================================================================== --- /dev/null +++ b/utils/apparmor/parser.py @@ -0,0 +1,76 @@ +#!/usr/bin/env python +# ---------------------------------------------------------------------- +# Copyright (C) 2013 Kshitij Gupta <[email protected]> +# Copyright (C) 2014 Christian Boltz <[email protected]> +# Copyright (C) 2014 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. +# +# ---------------------------------------------------------------------- + +from apparmor.regex import RE_PROFILE_CAP +from apparmor.common import AppArmorBug, AppArmorException +from apparmor.rule.capability import CapabilityRule +import re + +# setup module translations +from apparmor.translations import init_translation +_ = init_translation() + +def parse_capability(rawrule): + '''parse rawrule and return CapabilityRule''' + + matches = RE_PROFILE_CAP.search(rawrule) + if not matches: + raise AppArmorException(_("Invalid capability rule '%s'") % rawrule) + + rawrule = rawrule.strip() + + audit, deny, allow_keyword, inlinecomment = parse_modifiers(matches) + + capability = [] + + if matches.group('capability'): + capability = matches.group('capability').strip() + capability = re.split("[ \t]+", capability) + else: + capability = CapabilityRule.ALL + + return CapabilityRule(capability, audit=audit, deny=deny, + allow_keyword=allow_keyword, + comment=inlinecomment, raw_rule=rawrule) + +def parse_modifiers(matches): + '''returns audit, deny, allow_keyword and comment from the matches object + - audit, deny and allow_keyword are True/False + - comment is the comment with a leading space''' + audit = False + if matches.group('audit'): + audit = True + + deny = False + allow_keyword = False + + allowstr = matches.group('allow') + if allowstr: + if allowstr.strip() == 'allow': + allow_keyword = True + elif allowstr.strip() == 'deny': + deny = True + else: + raise AppArmorBug("Invalid allow/deny keyword %s" % allowstr) + + comment = '' + if matches.group('comment'): + # include a space so that we don't need to add it everywhere when writing the rule + comment = ' %s' % matches.group('comment') + + return (audit, deny, allow_keyword, comment) +
=== modified file 'utils/apparmor/aa.py'
--- a/utils/apparmor/aa.py 2014-11-27 22:20:26 +0000
+++ b/utils/apparmor/aa.py 2014-11-29 08:33:10 +0000
@@ -52,6 +52,9 @@
import apparmor.rules as aarules
+from apparmor.rule.capability import CapabilityRuleset, CapabilityRule
+import apparmor.parser as aaparser
+
from apparmor.yasti import SendDataToYast, GetDataFromYast, shutdown_yast
# setup module translations
@@ -89,13 +92,19 @@
# To store the globs entered by users so they can be provided again
user_globs = []
-# The key for representing bare rules such as "capability," or "file,"
+# The key for representing bare "file," rules
ALL = '\0ALL'
## Variables used under logprof
### Were our
t = hasher() # dict()
transitions = hasher()
+
+# keys used in aa[profile][hat]:
+# a) rules (as dict): alias, change_profile, include, lvar, rlimit
+# b) rules (as hasher): allow, deny
+# c) one for each rule class
+# d) other: declared, external, flags, name, profile
aa = hasher() # Profiles originally in sd, replace by aa
original_aa = hasher()
extras = hasher() # Inactive profiles from extras
@@ -1546,13 +1555,14 @@
for hat in hats:
for capability in sorted(log_dict[aamode][profile][hat]['capability'].keys()):
# skip if capability already in profile
- if profile_known_capability(aa[profile][hat], capability):
+ capability_obj = CapabilityRule(capability)
+ if is_known_rule(aa[profile][hat], 'capability', capability_obj):
continue
# Load variables? Don't think so.
severity = sev_db.rank('CAP_%s' % capability)
default_option = 1
options = []
- newincludes = match_cap_includes(aa[profile][hat], capability)
+ newincludes = match_includes(aa[profile][hat], 'capability', capability_obj)
q = aaui.PromptQuestion()
if newincludes:
@@ -1625,16 +1635,18 @@
if deleted:
aaui.UI_Info(_('Deleted %s previous matching profile entries.') % deleted)
- aa[profile][hat]['allow']['capability'][capability]['set'] = True
- aa[profile][hat]['allow']['capability'][capability]['audit'] = audit_toggle
+ else:
+ capability_obj = CapabilityRule(capability, audit=audit)
+ aa[profile][hat]['capability'].add(capability_obj)
+ aaui.UI_Info(_('Adding capability %s to profile.') % capability)
changed[profile] = True
- aaui.UI_Info(_('Adding capability %s to profile.') % capability)
done = True
elif ans == 'CMD_DENY':
- aa[profile][hat]['deny']['capability'][capability]['set'] = True
+ capability_obj = CapabilityRule(capability, audit=audit, deny=True)
+ aa[profile][hat]['capability'].add(capability_obj)
changed[profile] = True
aaui.UI_Info(_('Denying capability %s to profile.') % capability)
@@ -2119,20 +2131,6 @@
deleted += 1
return deleted
-def delete_cap_duplicates(profilecaps, inccaps):
- deleted = []
- if profilecaps and inccaps:
- for capname in profilecaps.keys():
- # XXX The presence of a bare capability rule ("capability,") should
- # cause more specific capability rules
- # ("capability audit_control,") to be deleted
- if inccaps[capname].get('set', False) == 1:
- deleted.append(capname)
- for capname in deleted:
- profilecaps.pop(capname)
-
- return len(deleted)
-
def delete_path_duplicates(profile, incname, allow):
deleted = []
for entry in profile[allow]['path'].keys():
@@ -2159,9 +2157,7 @@
deleted += delete_net_duplicates(profile['deny']['netdomain'], include[incname][incname]['deny']['netdomain'])
- deleted += delete_cap_duplicates(profile['allow']['capability'], include[incname][incname]['allow']['capability'])
-
- deleted += delete_cap_duplicates(profile['deny']['capability'], include[incname][incname]['deny']['capability'])
+ deleted += profile['capability'].delete_duplicates(include[incname][incname]['capability'])
deleted += delete_path_duplicates(profile, incname, 'allow')
deleted += delete_path_duplicates(profile, incname, 'deny')
@@ -2171,9 +2167,7 @@
deleted += delete_net_duplicates(profile['deny']['netdomain'], filelist[incname][incname]['deny']['netdomain'])
- deleted += delete_cap_duplicates(profile['allow']['capability'], filelist[incname][incname]['allow']['capability'])
-
- deleted += delete_cap_duplicates(profile['deny']['capability'], filelist[incname][incname]['deny']['capability'])
+ deleted += profile['capability'].delete_duplicates(filelist[incname][incname]['capability'])
deleted += delete_path_duplicates(profile, incname, 'allow')
deleted += delete_path_duplicates(profile, incname, 'deny')
@@ -2201,10 +2195,16 @@
return False
-def match_cap_includes(profile, cap):
+def match_cap_includes(profile, capability):
+ # still used by aa-mergeprof
+ capability_obj = CapabilityRule(capability)
+ return match_includes(profile, 'capability', capability_obj)
+
+def match_includes(profile, rule_type, rule_obj):
newincludes = []
for incname in include.keys():
- if valid_include(profile, incname) and include[incname][incname]['allow']['capability'][cap].get('set', False) == 1:
+ # XXX type check should go away once we init all profiles correctly
+ if valid_include(profile, incname) and include[incname][incname].get(rule_type, False) and include[incname][incname][rule_type].is_covered(rule_obj):
newincludes.append(incname)
return newincludes
@@ -2512,10 +2512,11 @@
log_dict[aamode][profile][hat]['path'][path] = mode
- for capability in prelog[aamode][profile][hat]['capability'].keys():
+ for cap in prelog[aamode][profile][hat]['capability'].keys():
# If capability not already in profile
- if not aa[profile][hat]['allow']['capability'][capability].get('set', False):
- log_dict[aamode][profile][hat]['capability'][capability] = True
+ # XXX remove first check when we have proper profile initialisation
+ if aa[profile][hat].get('capability', False) and not aa[profile][hat]['capability'].is_covered(CapabilityRule(cap)):
+ log_dict[aamode][profile][hat]['capability'][cap] = True
nd = prelog[aamode][profile][hat]['netdomain']
for family in nd.keys():
@@ -2702,6 +2703,10 @@
profile_data[profile][profile]['repo']['url'] = repo_data['url']
profile_data[profile][profile]['repo']['user'] = repo_data['user']
+ # init rule classes (if not done yet)
+ if not profile_data[profile][hat].get('capability', False):
+ profile_data[profile][hat]['capability'] = CapabilityRuleset()
+
elif RE_PROFILE_END.search(line):
# If profile ends and we're not in one
if not profile:
@@ -2717,21 +2722,14 @@
initial_comment = ''
elif RE_PROFILE_CAP.search(line):
- matches = RE_PROFILE_CAP.search(line)
-
if not profile:
raise AppArmorException(_('Syntax Error: Unexpected capability entry found in file: %(file)s line: %(line)s') % { 'file': file, 'line': lineno + 1 })
- audit, allow, allow_keyword, comment = parse_audit_allow(matches)
- # TODO: honor allow_keyword and comment
-
- capability = ALL
- if matches.group('capability'):
- capability = matches.group('capability').strip()
- # TODO: can contain more than one capability- split it?
-
- profile_data[profile][hat][allow]['capability'][capability]['set'] = True
- profile_data[profile][hat][allow]['capability'][capability]['audit'] = audit
+ # init rule class (if not done yet)
+ if not profile_data[profile][hat].get('capability', False):
+ profile_data[profile][hat]['capability'] = CapabilityRuleset()
+
+ profile_data[profile][hat]['capability'].add(aaparser.parse_capability(line))
elif RE_PROFILE_LINK.search(line):
matches = RE_PROFILE_LINK.search(line).groups()
@@ -2840,7 +2838,7 @@
if not profile:
raise AppArmorException(_('Syntax Error: Unexpected bare file rule found in file: %(file)s line: %(line)s') % { 'file': file, 'line': lineno + 1 })
- audit, allow, allow_keyword, comment = parse_audit_allow(matches)
+ audit, allow, allow_keyword, comment = aaparser.parse_modifiers(matches)
# TODO: honor allow_keyword and comment
mode = apparmor.aamode.AA_BARE_FILE_MODE
@@ -3179,26 +3177,6 @@
return profile_data
-def parse_audit_allow(matches):
- audit = False
- if matches.group('audit'):
- audit = True
-
- allow = 'allow'
- allow_keyword = False
- if matches.group('allow'):
- allow = matches.group('allow').strip()
- allow_keyword = True
- if allow != 'allow' and allow != 'deny': # should never happen
- raise AppArmorException(_("Invalid allow/deny keyword %s" % allow))
-
- comment = ''
- if matches.group('comment'):
- # include a space so that we don't need to add it everywhere when writing the rule
- comment = ' %s' % matches.group('comment')
-
- return (audit, allow, allow_keyword, comment)
-
# RE_DBUS_ENTRY = re.compile('^dbus\s*()?,\s*$')
# use stuff like '(?P<action>(send|write|w|receive|read|r|rw))'
@@ -3371,29 +3349,10 @@
def write_list_vars(prof_data, depth):
return write_pair(prof_data, depth, '', 'lvar', '', ' = ', '', var_transform)
-def write_cap_rules(prof_data, depth, allow):
- pre = ' ' * depth
- data = []
- allowstr = set_allow_str(allow)
-
- if prof_data[allow].get('capability', False):
- for cap in sorted(prof_data[allow]['capability'].keys()):
- audit = ''
- if prof_data[allow]['capability'][cap].get('audit', False):
- audit = 'audit '
- if prof_data[allow]['capability'][cap].get('set', False):
- if cap == ALL:
- data.append('%s%s%scapability,' % (pre, audit, allowstr))
- else:
- data.append('%s%s%scapability %s,' % (pre, audit, allowstr, cap))
- data.append('')
-
- return data
-
def write_capabilities(prof_data, depth):
- #data = write_single(prof_data, depth, '', 'set_capability', 'set capability ', ',')
- data = write_cap_rules(prof_data, depth, 'deny')
- data += write_cap_rules(prof_data, depth, 'allow')
+ data = []
+ if prof_data.get('capability', False):
+ data = prof_data['capability'].get_clean(depth)
return data
def write_net_rules(prof_data, depth, allow):
@@ -3823,10 +3782,14 @@
depth = len(line) - len(line.lstrip())
data += write_methods[segs](prof_data, int(depth / 2))
segments[segs] = False
+ # delete rules from prof_data to avoid duplication (they are in data now)
if prof_data['allow'].get(segs, False):
prof_data['allow'].pop(segs)
if prof_data['deny'].get(segs, False):
prof_data['deny'].pop(segs)
+ if prof_data.get(segs, False):
+ t = type(prof_data[segs])
+ prof_data[segs] = t()
return data
#data.append('reading prof')
@@ -3887,16 +3850,20 @@
if profile:
depth = int(len(line) - len(line.lstrip()) / 2) + 1
- # first write sections that were modified (and remove them from write_prof_data)
+ # first write sections that were modified
#for segs in write_methods.keys():
for segs in default_write_order:
if segments[segs]:
data += write_methods[segs](write_prof_data[name], depth)
segments[segs] = False
+ # delete rules from write_prof_data to avoid duplication (they are in data now)
if write_prof_data[name]['allow'].get(segs, False):
write_prof_data[name]['allow'].pop(segs)
if write_prof_data[name]['deny'].get(segs, False):
write_prof_data[name]['deny'].pop(segs)
+ if write_prof_data[name].get(segs, False):
+ t = type(write_prof_data[name][segs])
+ write_prof_data[name][segs] = t()
# then write everything else
for segs in default_write_order:
@@ -3938,34 +3905,13 @@
profile = None
elif RE_PROFILE_CAP.search(line):
- matches = RE_PROFILE_CAP.search(line).groups()
- audit = False
- if matches[0]:
- audit = matches[0]
-
- allow = 'allow'
- if matches[1] and matches[1].strip() == 'deny':
- allow = 'deny'
-
- capability = ALL
- if matches[2]:
- capability = matches[2].strip()
-
- if not write_prof_data[hat][allow]['capability'][capability].get('set', False):
- correct = False
- if not write_prof_data[hat][allow]['capability'][capability].get(audit, False) == audit:
- correct = False
-
- if correct:
+ cap = aaparser.parse_capability(line)
+ if write_prof_data[hat]['capability'].is_covered(cap, True, True):
if not segments['capability'] and True in segments.values():
data += write_prior_segments(write_prof_data[name], segments, line)
segments['capability'] = True
- write_prof_data[hat][allow]['capability'].pop(capability)
+ write_prof_data[hat]['capability'].delete(cap)
data.append(line)
-
- #write_prof_data[hat][allow]['capability'][capability].pop(audit)
-
- #Remove this line
else:
# To-Do
pass
@@ -4341,20 +4287,18 @@
return 0
-def profile_known_capability(profile, capname):
- if profile['deny']['capability'][capname].get('set', False):
- return -1
-
- if profile['allow']['capability'][capname].get('set', False):
- return 1
+def is_known_rule(profile, rule_type, rule_obj):
+ # XXX get rid of get() checks after we have a proper function to initialize a profile
+ if profile.get(rule_type, False):
+ if profile[rule_type].is_covered(rule_obj, False):
+ return True
for incname in profile['include'].keys():
- if include[incname][incname]['deny']['capability'][capname].get('set', False):
- return -1
- if include[incname][incname]['allow']['capability'][capname].get('set', False):
- return 1
+ if include[incname][incname].get(rule_type, False):
+ if include[incname][incname][rule_type].is_covered(rule_obj, False):
+ return True
- return 0
+ return False
def profile_known_network(profile, family, sock_type):
if netrules_access_check(profile['deny']['netdomain'], family, sock_type):
=== modified file 'utils/apparmor/cleanprofile.py'
--- a/utils/apparmor/cleanprofile.py 2014-09-05 21:21:00 +0000
+++ b/utils/apparmor/cleanprofile.py 2014-11-29 08:24:31 +0000
@@ -65,8 +65,8 @@
deleted += apparmor.aa.delete_duplicates(self.other.aa[program][hat], inc)
#Clean the duplicates of caps in other profile
- deleted += delete_cap_duplicates(self.profile.aa[program][hat]['allow']['capability'], self.other.aa[program][hat]['allow']['capability'], self.same_file)
- deleted += delete_cap_duplicates(self.profile.aa[program][hat]['deny']['capability'], self.other.aa[program][hat]['deny']['capability'], self.same_file)
+ if self.same_file:
+ deleted += self.other.aa[program][hat]['capability'].delete_duplicates(self.profile.aa[program][hat]['capability'])
#Clean the duplicates of path in other profile
deleted += delete_path_duplicates(self.profile.aa[program][hat], self.other.aa[program][hat], 'allow', self.same_file)
@@ -108,17 +108,6 @@
return len(deleted)
-def delete_cap_duplicates(profilecaps, profilecaps_other, same_profile=True):
- deleted = []
- if profilecaps and profilecaps_other and not same_profile:
- for capname in profilecaps.keys():
- if profilecaps_other[capname].get('set', False):
- deleted.append(capname)
- for capname in deleted:
- profilecaps_other.pop(capname)
-
- return len(deleted)
-
def delete_net_duplicates(netrules, netrules_other, same_profile=True):
deleted = 0
hasher_obj = apparmor.aa.hasher()
=== added file 'utils/apparmor/parser.py'
--- a/utils/apparmor/parser.py 1970-01-01 00:00:00 +0000
+++ b/utils/apparmor/parser.py 2014-11-29 08:24:32 +0000
@@ -0,0 +1,76 @@
+#!/usr/bin/env python
+# ----------------------------------------------------------------------
+# Copyright (C) 2013 Kshitij Gupta <[email protected]>
+# Copyright (C) 2014 Christian Boltz <[email protected]>
+# Copyright (C) 2014 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.
+#
+# ----------------------------------------------------------------------
+
+from apparmor.regex import RE_PROFILE_CAP
+from apparmor.common import AppArmorBug, AppArmorException
+from apparmor.rule.capability import CapabilityRule
+import re
+
+# setup module translations
+from apparmor.translations import init_translation
+_ = init_translation()
+
+def parse_capability(rawrule):
+ '''parse rawrule and return CapabilityRule'''
+
+ matches = RE_PROFILE_CAP.search(rawrule)
+ if not matches:
+ raise AppArmorException(_("Invalid capability rule '%s'") % rawrule)
+
+ rawrule = rawrule.strip()
+
+ audit, deny, allow_keyword, inlinecomment = parse_modifiers(matches)
+
+ capability = []
+
+ if matches.group('capability'):
+ capability = matches.group('capability').strip()
+ capability = re.split("[ \t]+", capability)
+ else:
+ capability = CapabilityRule.ALL
+
+ return CapabilityRule(capability, audit=audit, deny=deny,
+ allow_keyword=allow_keyword,
+ comment=inlinecomment, raw_rule=rawrule)
+
+def parse_modifiers(matches):
+ '''returns audit, deny, allow_keyword and comment from the matches object
+ - audit, deny and allow_keyword are True/False
+ - comment is the comment with a leading space'''
+ audit = False
+ if matches.group('audit'):
+ audit = True
+
+ deny = False
+ allow_keyword = False
+
+ allowstr = matches.group('allow')
+ if allowstr:
+ if allowstr.strip() == 'allow':
+ allow_keyword = True
+ elif allowstr.strip() == 'deny':
+ deny = True
+ else:
+ raise AppArmorBug("Invalid allow/deny keyword %s" % allowstr)
+
+ comment = ''
+ if matches.group('comment'):
+ # include a space so that we don't need to add it everywhere when writing the rule
+ comment = ' %s' % matches.group('comment')
+
+ return (audit, deny, allow_keyword, comment)
+
=== added directory 'utils/apparmor/rule'
=== added file 'utils/apparmor/rule/__init__.py'
--- a/utils/apparmor/rule/__init__.py 1970-01-01 00:00:00 +0000
+++ b/utils/apparmor/rule/__init__.py 2014-11-29 09:25:42 +0000
@@ -0,0 +1,189 @@
+# ----------------------------------------------------------------------
+# Copyright (C) 2013 Kshitij Gupta <[email protected]>
+# Copyright (C) 2014 Christian Boltz <[email protected]>
+#
+# 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.
+#
+# ----------------------------------------------------------------------
+
+from apparmor.common import AppArmorBug
+
+# setup module translations
+from apparmor.translations import init_translation
+_ = init_translation()
+
+
+class BaseRule(object):
+ '''Base class to handle and store a single rule'''
+
+ def __init__(self, audit=False, deny=False, allow_keyword=False,
+ comment='', log_event=None, raw_rule=''):
+ '''initialize variables needed by all rule types'''
+ self.audit = audit
+ self.deny = deny
+ self.allow_keyword = allow_keyword
+
+ self.comment = comment
+
+ self.raw_rule = raw_rule.strip() if raw_rule else None
+ self.log_event = log_event
+
+ def get_raw(self, depth=0):
+ '''return raw rule (with original formatting, and leading whitespace in the depth parameter)'''
+ if self.raw_rule:
+ return '%s%s' % (' ' * depth, self.raw_rule)
+ else:
+ return self.get_clean(depth)
+
+ def is_equal(self, rule_obj, strict=False):
+ '''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:
+ return False
+
+ if strict and (
+ self.allow_keyword != rule_obj.allow_keyword
+ or self.comment != rule_obj.comment
+ or self.raw_rule != rule_obj.raw_rule
+ ):
+ return False
+
+ return self.is_equal_localvars(rule_obj)
+
+ def modifiers_str(self):
+ '''return the allow/deny and audit keyword as string, including whitespace'''
+
+ if self.audit:
+ auditstr = 'audit '
+ else:
+ auditstr = ''
+
+ if self.deny:
+ allowstr = 'deny '
+ elif self.allow_keyword:
+ allowstr = 'allow '
+ else:
+ allowstr = ''
+
+ return '%s%s' % (auditstr, allowstr)
+
+
+class BaseRuleset(object):
+ '''Base class to handle and store a collection of rules'''
+
+ # decides if the (G)lob and Glob w/ (E)xt options are displayed
+ can_glob = True
+ can_glob_ext = False
+
+ def __init__(self):
+ '''initialize variables needed by all ruleset types
+ Do not override in child class unless really needed - override _init_vars() instead'''
+ self.rules = []
+ self._init_vars()
+
+ def _init_vars(self):
+ '''called by __init__() and delete_all_rules() - override in child class to initialize more variables'''
+ pass
+
+ def add(self, rule):
+ '''add a rule object'''
+ self.rules.append(rule)
+
+ def get_raw(self, depth=0):
+ '''return all raw rules (if possible/not modified in their original formatting).
+ Returns an array of lines, with depth * leading whitespace'''
+
+ data = []
+ for rule in self.rules:
+ data.append(rule.get_raw(depth))
+
+ if data:
+ data.append('')
+
+ return data
+
+ def get_clean(self, depth=0):
+ '''return all rules (in clean/default formatting)
+ Returns an array of lines, with depth * leading whitespace'''
+
+ allow_rules = []
+ deny_rules = []
+
+ for rule in self.rules:
+ if rule.deny:
+ deny_rules.append(rule.get_clean(depth))
+ else:
+ allow_rules.append(rule.get_clean(depth))
+
+ allow_rules.sort()
+ deny_rules.sort()
+
+ cleandata = []
+
+ if deny_rules:
+ cleandata += deny_rules
+ cleandata.append('')
+
+ if allow_rules:
+ cleandata += allow_rules
+ cleandata.append('')
+
+ return cleandata
+
+ def is_covered(self, rule, check_allow_deny=True, check_audit=False):
+ '''return True if rule is covered by existing rules, otherwise False'''
+
+ for r in self.rules:
+ if r.is_covered(rule, check_allow_deny, check_audit):
+ return True
+
+ return False
+
+# def is_log_covered(self, parsed_log_event, check_allow_deny=True, check_audit=False):
+# '''return True if parsed_log_event is covered by existing rules, otherwise False'''
+#
+# rule_obj = self.new_rule()
+# rule_obj.set_log(parsed_log_event)
+#
+# return self.is_covered(rule_obj, check_allow_deny, check_audit)
+
+ def delete(self, rule):
+ '''Delete rule from rules'''
+
+ rule_to_delete = False
+ i = 0
+ for r in self.rules:
+ if r.is_equal(rule):
+ rule_to_delete = True
+ break
+ i = i + 1
+
+ if rule_to_delete:
+ self.rules.pop(i)
+ else:
+ raise AppArmorBug('Attempt to delete non-existing rule %s' % rule.get_raw(0))
+
+ def delete_duplicates(self, include_rules):
+ '''Delete duplicate rules.
+ include_rules must be a *_rules object'''
+ deleted = []
+ if include_rules: # avoid breakage until we have a proper function to ensure all profiles contain all *_rules objects
+ for rule in self.rules:
+ if include_rules.is_covered(rule, True, True):
+ self.delete(rule)
+ deleted.append(rule)
+
+ return len(deleted)
+
+ def get_glob_ext(self, path_or_rule):
+ '''returns the next possible glob with extension (for file rules only).
+ For all other rule types, raise an exception'''
+ raise AppArmorBug("get_glob_ext is not available for this rule type!")
=== added file 'utils/apparmor/rule/capability.py'
--- a/utils/apparmor/rule/capability.py 1970-01-01 00:00:00 +0000
+++ b/utils/apparmor/rule/capability.py 2014-11-29 10:22:58 +0000
@@ -0,0 +1,119 @@
+#!/usr/bin/env python
+# ----------------------------------------------------------------------
+# Copyright (C) 2013 Kshitij Gupta <[email protected]>
+# Copyright (C) 2014 Christian Boltz <[email protected]>
+#
+# 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.
+#
+# ----------------------------------------------------------------------
+
+from apparmor.common import AppArmorBug
+from apparmor.rule import BaseRule, BaseRuleset
+
+# setup module translations
+from apparmor.translations import init_translation
+_ = init_translation()
+
+
+class CapabilityRule(BaseRule):
+ '''Class to handle and store a single capability rule'''
+
+ # Nothing external should reference this class, all external users
+ # should reference the class field CapabilityRule.ALL
+ class __CapabilityAll(object):
+ pass
+
+ ALL = __CapabilityAll
+
+ def __init__(self, cap_list, audit=False, deny=False, allow_keyword=False,
+ comment='', log_event=None, raw_rule=None):
+
+ super(CapabilityRule, self).__init__(audit=audit, deny=deny,
+ allow_keyword=allow_keyword,
+ comment=comment,
+ log_event=log_event,
+ raw_rule=raw_rule)
+ # Because we support having multiple caps in one rule,
+ # initializer needs to accept a list of caps.
+ self.all_caps = False
+ if cap_list == CapabilityRule.ALL:
+ self.all_caps = True
+ self.capability = set()
+ else:
+ if type(cap_list) == str:
+ self.capability = {cap_list}
+ elif type(cap_list) == list and len(cap_list) > 0:
+ self.capability = set(cap_list)
+ else:
+ raise AppArmorBug('Passed unknown object to CapabilityRule: %s' % str(cap_list))
+ # make sure none of the cap_list arguments are blank, in
+ # case we decide to return one cap per output line
+ for cap in self.capability:
+ if len(cap.strip()) == 0:
+ raise AppArmorBug('Passed empty capability to CapabilityRule: %s' % str(cap_list))
+
+ def get_clean(self, depth=0):
+ '''return rule (in clean/default formatting)'''
+
+ space = ' ' * depth
+ if self.all_caps:
+ return('%s%scapability,%s' % (space, self.modifiers_str(), self.comment))
+ else:
+ caps = ' '.join(self.capability).strip() # XXX return multiple lines, one for each capability, instead?
+ if caps:
+ return('%s%scapability %s,%s' % (space, self.modifiers_str(), ' '.join(sorted(self.capability)), self.comment))
+ else:
+ raise AppArmorBug("Empty capability rule")
+
+ def is_covered(self, rule_obj, check_allow_deny=True, check_audit=False):
+ '''check if rule_obj is covered by this rule object'''
+
+ if not type(rule_obj) == CapabilityRule:
+ raise AppArmorBug('Passes non-capability rule: %s' % str(rule_obj))
+
+ if check_allow_deny and self.deny != rule_obj.deny:
+ return False
+
+ if not rule_obj.capability and not rule_obj.all_caps:
+ raise AppArmorBug('No capability specified')
+
+ if not self.all_caps:
+ if rule_obj.all_caps:
+ return False
+ if not rule_obj.capability.issubset(self.capability):
+ return False
+
+ if check_audit and rule_obj.audit != self.audit:
+ return False
+
+ if rule_obj.audit and not self.audit:
+ return False
+
+ # still here? -> then it is covered
+ return True
+
+ def is_equal_localvars(self, rule_obj):
+ '''compare if rule-specific variables are equal'''
+
+ if not type(rule_obj) == CapabilityRule:
+ raise AppArmorBug('Passes non-capability rule: %s' % str(rule_obj))
+
+ if (self.capability != rule_obj.capability
+ or self.all_caps != rule_obj.all_caps):
+ return False
+
+ return True
+
+class CapabilityRuleset(BaseRuleset):
+ '''Class to handle and store a collection of capability rules'''
+
+ def get_glob(self, path_or_rule):
+ '''Return the next possible glob. For capability rules, that's always "capability," (all capabilities)'''
+ return 'capability,'
=== added file 'utils/test/test-capability.py'
--- a/utils/test/test-capability.py 1970-01-01 00:00:00 +0000
+++ b/utils/test/test-capability.py 2014-11-29 10:20:52 +0000
@@ -0,0 +1,866 @@
+#!/usr/bin/env python
+# ----------------------------------------------------------------------
+# Copyright (C) 2014 Christian Boltz <[email protected]>
+#
+# 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.rule.capability import CapabilityRule, CapabilityRuleset
+from apparmor.parser import parse_capability, parse_modifiers
+from apparmor.rule import BaseRule
+from apparmor.common import AppArmorException, AppArmorBug, hasher
+from apparmor.logparser import ReadLog
+
+import re
+
+# --- tests for single CapabilityRule --- #
+
+class CapabilityTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+
+ def _compare_obj_with_rawrule(self, rawrule, expected):
+
+ obj = parse_capability(rawrule)
+
+ self.assertEqual(rawrule.strip(), obj.raw_rule)
+
+ self._compare_obj(obj, expected)
+
+ def _compare_obj(self, obj, expected):
+ self.assertEqual(expected['allow_keyword'], obj.allow_keyword)
+ self.assertEqual(expected['audit'], obj.audit)
+ self.assertEqual(expected['capability'], obj.capability)
+ self.assertEqual(expected['all_caps'], obj.all_caps)
+ self.assertEqual(expected['deny'], obj.deny)
+ self.assertEqual(expected['inlinecomment'], obj.comment)
+
+ def test_cap_allow_all(self):
+ self._compare_obj_with_rawrule("capability,", {
+ 'allow_keyword': False,
+ 'deny': False,
+ 'audit': False,
+ 'capability': set(),
+ 'all_caps': True,
+ 'inlinecomment': "",
+ })
+
+ def test_cap_allow_sys_admin(self):
+ self._compare_obj_with_rawrule("capability sys_admin,", {
+ 'allow_keyword': False,
+ 'deny': False,
+ 'audit': False,
+ 'capability': {'sys_admin'},
+ 'all_caps': False,
+ 'inlinecomment': "",
+ })
+
+ def test_cap_deny_sys_admin(self):
+ self._compare_obj_with_rawrule(" deny capability sys_admin, # some comment", {
+ 'allow_keyword': False,
+ 'deny': True,
+ 'audit': False,
+ 'capability': {'sys_admin'},
+ 'all_caps': False,
+ 'inlinecomment': " # some comment",
+ })
+
+ def test_cap_multi(self):
+ self._compare_obj_with_rawrule("capability sys_admin dac_override,", {
+ 'allow_keyword': False,
+ 'deny': False,
+ 'audit': False,
+ 'capability': {'sys_admin', 'dac_override'},
+ 'all_caps': False,
+ 'inlinecomment': "",
+ })
+
+ # Template for test_cap_* functions
+ # def test_cap_(self):
+ # self._compare_obj_with_rawrule("capability,", {
+ # 'allow_keyword': False,
+ # 'deny': False,
+ # 'audit': False,
+ # 'capability': set(), # (or {'foo'} if not empty)
+ # 'all_caps': False,
+ # 'inlinecomment': "",
+ # })
+
+ def test_cap_from_log(self):
+ parser = ReadLog('', '', '', '', '')
+ event = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname="net_raw"'
+
+ parsed_event = parser.parse_event(event)
+
+ self.assertEqual(parsed_event, {
+ 'request_mask': set(),
+ 'denied_mask': set(),
+ 'magic_token': 0,
+ 'parent': 0,
+ 'profile': '/bin/ping',
+ 'operation': 'capable',
+ 'resource': None,
+ 'info': None,
+ 'aamode': 'PERMITTING',
+ 'time': 1415403814,
+ 'active_hat': None,
+ 'pid': 15454,
+ 'task': 0,
+ 'attr': None,
+ 'name2': None,
+ 'name': 'net_raw'
+ })
+
+ obj = CapabilityRule(parsed_event['name'], log_event=parsed_event)
+
+ self._compare_obj(obj, {
+ 'allow_keyword': False,
+ 'deny': False,
+ 'audit': False,
+ 'capability': {'net_raw'},
+ 'all_caps': False,
+ 'inlinecomment': "",
+ })
+
+ self.assertEqual(obj.get_raw(1), ' capability net_raw,')
+
+# def test_cap_from_invalid_log(self):
+# parser = ReadLog('', '', '', '', '')
+# # invalid log entry, name= should contain the capability name
+# event = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname=""'
+#
+# parsed_event = parser.parse_event(event)
+#
+# obj = CapabilityRule()
+#
+# with self.assertRaises(AppArmorBug):
+# obj.set_log(parsed_event)
+#
+# with self.assertRaises(AppArmorBug):
+# obj.get_raw(1)
+#
+# def test_cap_from_non_cap_log(self):
+# parser = ReadLog('', '', '', '', '')
+# # log entry for different rule type
+# event = 'type=AVC msg=audit(1415403814.973:667): apparmor="ALLOWED" operation="setsockopt" profile="/home/sys-tmp/ping" pid=15454 comm="ping" lport=1 family="inet" sock_type="raw" protocol=1'
+#
+# parsed_event = parser.parse_event(event)
+#
+# obj = CapabilityRule()
+#
+# with self.assertRaises(AppArmorBug):
+# obj.set_log(parsed_event)
+#
+# with self.assertRaises(AppArmorBug):
+# obj.get_raw(1)
+
+ def test_cap_from_init_01(self):
+ obj = CapabilityRule('chown')
+
+ self._compare_obj(obj, {
+ 'allow_keyword': False,
+ 'deny': False,
+ 'audit': False,
+ 'capability': {'chown'},
+ 'all_caps': False,
+ 'inlinecomment': "",
+ })
+
+ def test_cap_from_init_02(self):
+ obj = CapabilityRule(['chown'])
+
+ self._compare_obj(obj, {
+ 'allow_keyword': False,
+ 'deny': False,
+ 'audit': False,
+ 'capability': {'chown'},
+ 'all_caps': False,
+ 'inlinecomment': "",
+ })
+
+ def test_cap_from_init_03(self):
+ obj = CapabilityRule('chown', audit=True, deny=True)
+
+ self._compare_obj(obj, {
+ 'allow_keyword': False,
+ 'deny': True,
+ 'audit': True,
+ 'capability': {'chown'},
+ 'all_caps': False,
+ 'inlinecomment': "",
+ })
+
+ def test_cap_from_init_04(self):
+ obj = CapabilityRule(['chown', 'fsetid'], deny=True)
+
+ self._compare_obj(obj, {
+ 'allow_keyword': False,
+ 'deny': True,
+ 'audit': False,
+ 'capability': {'chown', 'fsetid'},
+ 'all_caps': False,
+ 'inlinecomment': "",
+ })
+
+
+class InvalidCapabilityTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+
+ # XXX - these tests more properly belong in a test suite for
+ # XXX - parser.py::parse_capability()
+ def _check_invalid_rawrule(self, rawrule):
+ obj = None
+ with self.assertRaises(AppArmorException):
+ obj = CapabilityRule(parse_capability(rawrule))
+
+ self.assertIsNone(obj, 'CapbilityRule handed back an object unexpectedly')
+
+ def test_invalid_cap_missing_comma(self):
+ self._check_invalid_rawrule('capability') # missing comma
+
+ def test_invalid_cap_non_CapabilityRule(self):
+ self._check_invalid_rawrule('network,') # not a capability rule
+
+ # XXX - this test more properly belongs in a test suite for
+ # XXX - parser.py::parse_modifiers()
+ def test_parse_modifiers_invalid(self):
+ regex = re.compile('^\s*(?P<audit>audit\s+)?(?P<allow>allow\s+|deny\s+|invalid\s+)?')
+ matches = regex.search('audit invalid ')
+
+ with self.assertRaises(AppArmorBug):
+ parse_modifiers(matches)
+
+ def test_empty_cap_set(self):
+ obj = CapabilityRule('chown')
+ obj.capability.clear()
+ # no capability set, and ALL not set
+ with self.assertRaises(AppArmorBug):
+ obj.get_clean(1)
+
+ def test_empty_cap_list(self):
+ with self.assertRaises(AppArmorBug):
+ CapabilityRule([])
+
+ def test_no_cap_list_arg(self):
+ with self.assertRaises(TypeError):
+ CapabilityRule()
+
+ def test_space_cap(self):
+ with self.assertRaises(AppArmorBug):
+ CapabilityRule(' ') # the whitespace capability ;-)
+
+ def test_space_list_1(self):
+ with self.assertRaises(AppArmorBug):
+ CapabilityRule([' ', ' ', ' ']) # the whitespace capability ;-)
+
+ def test_space_list_2(self):
+ with self.assertRaises(AppArmorBug):
+ CapabilityRule(['chown', ' ', 'setgid']) # includes the whitespace capability ;-)
+
+ def test_wrong_type_for_cap(self):
+ with self.assertRaises(AppArmorBug):
+ CapabilityRule(dict())
+
+
+class WriteCapabilityTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+
+ def _check_write_rule(self, rawrule, cleanrule):
+ obj = parse_capability(rawrule)
+ clean = obj.get_clean()
+ raw = obj.get_raw()
+
+ self.assertEqual(cleanrule.strip(), clean, 'unexpected clean rule')
+ self.assertEqual(rawrule.strip(), raw, 'unexpected raw rule')
+
+ def test_write_all(self):
+ self._check_write_rule(' capability , # foo ', 'capability, # foo')
+
+ def test_write_sys_admin(self):
+ self._check_write_rule(' audit capability sys_admin,', 'audit capability sys_admin,')
+
+ def test_write_sys_multi(self):
+ self._check_write_rule(' deny capability sys_admin audit_write,# foo bar', 'deny capability audit_write sys_admin, # foo bar')
+
+ def test_write_manually(self):
+ obj = CapabilityRule(['ptrace', 'audit_write'], allow_keyword=True)
+
+ expected = ' allow capability audit_write ptrace,'
+
+ self.assertEqual(expected, obj.get_clean(2), 'unexpected clean rule')
+ self.assertEqual(expected, obj.get_raw(2), 'unexpected raw rule')
+
+class CapabilityCoveredTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+
+ def _is_covered(self, obj, rule_to_test):
+ return obj.is_covered(parse_capability(rule_to_test))
+
+ def _is_covered_exact(self, obj, rule_to_test):
+ return obj.is_covered(parse_capability(rule_to_test), True, True)
+
+ def _is_equal(self, obj, rule_to_test, strict):
+ return obj.is_equal(parse_capability(rule_to_test), strict)
+
+ def test_covered_single(self):
+ obj = parse_capability('capability sys_admin,')
+
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin,'))
+
+ self.assertFalse(self._is_covered(obj, 'audit capability sys_admin,'))
+ self.assertFalse(self._is_covered(obj, 'audit capability,'))
+ self.assertFalse(self._is_covered(obj, 'capability chown,'))
+ self.assertFalse(self._is_covered(obj, 'capability,'))
+
+ def test_covered_audit(self):
+ obj = parse_capability('audit capability sys_admin,')
+
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin,'))
+ self.assertTrue(self._is_covered(obj, 'audit capability sys_admin,'))
+
+ self.assertFalse(self._is_covered(obj, 'audit capability,'))
+ self.assertFalse(self._is_covered(obj, 'capability chown,'))
+ self.assertFalse(self._is_covered(obj, 'capability,'))
+
+ def test_covered_check_audit(self):
+ obj = parse_capability('audit capability sys_admin,')
+
+ self.assertFalse(self._is_covered_exact(obj, 'capability sys_admin,'))
+ self.assertTrue(self._is_covered_exact(obj, 'audit capability sys_admin,'))
+
+ self.assertFalse(self._is_covered_exact(obj, 'audit capability,'))
+ self.assertFalse(self._is_covered_exact(obj, 'capability chown,'))
+ self.assertFalse(self._is_covered_exact(obj, 'capability,'))
+
+ def test_equal(self):
+ obj = parse_capability('capability sys_admin,')
+
+ self.assertTrue(self._is_equal(obj, 'capability sys_admin,', True))
+ self.assertFalse(self._is_equal(obj, 'allow capability sys_admin,', True))
+ self.assertFalse(self._is_equal(obj, 'allow capability sys_admin,', True))
+ self.assertFalse(self._is_equal(obj, 'audit capability sys_admin,', True))
+
+ self.assertTrue(self._is_equal(obj, 'capability sys_admin,', False))
+ self.assertTrue(self._is_equal(obj, 'allow capability sys_admin,', False))
+ self.assertFalse(self._is_equal(obj, 'audit capability sys_admin,', False))
+
+ def test_covered_multi(self):
+ obj = parse_capability('capability audit_write sys_admin,')
+
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin,'))
+ self.assertTrue(self._is_covered(obj, 'capability audit_write,'))
+ self.assertTrue(self._is_covered(obj, 'capability audit_write sys_admin,'))
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin audit_write,'))
+
+ self.assertFalse(self._is_covered(obj, 'audit capability,'))
+ self.assertFalse(self._is_covered(obj, 'capability chown,'))
+ self.assertFalse(self._is_covered(obj, 'capability,'))
+
+ def test_covered_all(self):
+ obj = parse_capability('capability,')
+
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin,'))
+ self.assertTrue(self._is_covered(obj, 'capability audit_write,'))
+ self.assertTrue(self._is_covered(obj, 'capability audit_write sys_admin,'))
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin audit_write,'))
+ self.assertTrue(self._is_covered(obj, 'capability,'))
+
+ self.assertFalse(self._is_covered(obj, 'audit capability,'))
+
+ def test_covered_deny(self):
+ obj = parse_capability('capability sys_admin,')
+
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin,'))
+
+ self.assertFalse(self._is_covered(obj, 'audit deny capability sys_admin,'))
+ self.assertFalse(self._is_covered(obj, 'deny capability sys_admin,'))
+ self.assertFalse(self._is_covered(obj, 'capability chown,'))
+ self.assertFalse(self._is_covered(obj, 'capability,'))
+
+ def test_covered_deny_2(self):
+ obj = parse_capability('deny capability sys_admin,')
+
+ self.assertTrue(self._is_covered(obj, 'deny capability sys_admin,'))
+
+ self.assertFalse(self._is_covered(obj, 'audit deny capability sys_admin,'))
+ self.assertFalse(self._is_covered(obj, 'capability sys_admin,'))
+ self.assertFalse(self._is_covered(obj, 'deny capability chown,'))
+ self.assertFalse(self._is_covered(obj, 'deny capability,'))
+
+ def test_invalid_is_covered(self):
+ obj = parse_capability('capability sys_admin,')
+
+ testobj = BaseRule() # different type
+
+ with self.assertRaises(AppArmorBug):
+ obj.is_covered(testobj)
+
+ def test_borked_obj_is_covered(self):
+ obj = parse_capability('capability sys_admin,')
+
+ testobj = CapabilityRule('chown')
+ testobj.capability.clear()
+
+ with self.assertRaises(AppArmorBug):
+ obj.is_covered(testobj)
+
+ def test_invalid_is_equal(self):
+ obj = parse_capability('capability sys_admin,')
+
+ testobj = BaseRule() # different type
+
+ with self.assertRaises(AppArmorBug):
+ obj.is_equal(testobj)
+
+ def test_empty_init(self):
+ # add to internal set instead of using .set_* (which overwrites the internal set) to make sure obj and obj2 use separate storage
+ obj = CapabilityRule('fsetid')
+ obj2 = CapabilityRule('fsetid')
+ obj.capability.add('sys_admin')
+ obj2.capability.add('ptrace')
+
+ self.assertTrue(self._is_covered(obj, 'capability sys_admin,'))
+ self.assertFalse(self._is_covered(obj, 'capability ptrace,'))
+ self.assertFalse(self._is_covered(obj2, 'capability sys_admin,'))
+ self.assertTrue(self._is_covered(obj2, 'capability ptrace,'))
+
+# --- tests for CapabilityRuleset --- #
+
+class CapabilityRulesTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+
+ def test_empty_ruleset(self):
+ ruleset = CapabilityRuleset()
+ ruleset_2 = CapabilityRuleset()
+ self.assertEqual([], ruleset.get_raw(2))
+ self.assertEqual([], ruleset.get_clean(2))
+ self.assertEqual([], ruleset_2.get_raw(2))
+ self.assertEqual([], ruleset_2.get_clean(2))
+
+ def test_ruleset_1(self):
+ ruleset = CapabilityRuleset()
+ rules = [
+ 'capability sys_admin,',
+ 'capability chown,',
+ ]
+
+ expected_raw = [
+ 'capability sys_admin,',
+ 'capability chown,',
+ '',
+ ]
+
+ expected_clean = [
+ 'capability chown,',
+ 'capability sys_admin,',
+ '',
+ ]
+
+ for rule in rules:
+ ruleset.add(parse_capability(rule))
+
+ self.assertEqual(expected_raw, ruleset.get_raw())
+ self.assertEqual(expected_clean, ruleset.get_clean())
+
+ def test_ruleset_2(self):
+ ruleset = CapabilityRuleset()
+ rules = [
+ 'capability chown,',
+ 'allow capability sys_admin,',
+ 'deny capability chgrp, # example comment',
+ ]
+
+ expected_raw = [
+ ' capability chown,',
+ ' allow capability sys_admin,',
+ ' deny capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' allow capability sys_admin,',
+ ' capability chown,',
+ '',
+ ]
+
+ for rule in rules:
+ ruleset.add(parse_capability(rule))
+
+ self.assertEqual(expected_raw, ruleset.get_raw(1))
+ self.assertEqual(expected_clean, ruleset.get_clean(1))
+
+ def test_ruleset_add(self):
+ rule = CapabilityRule('chgrp', comment=' # example comment')
+
+ ruleset = CapabilityRuleset()
+ ruleset.add(rule)
+
+ expected_raw = [
+ ' capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = expected_raw
+
+ self.assertEqual(expected_raw, ruleset.get_raw(1))
+ self.assertEqual(expected_clean, ruleset.get_clean(1))
+
+
+class CapabilityRulesCoveredTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+
+ self.ruleset = CapabilityRuleset()
+ rules = [
+ 'capability chown,',
+ 'capability setuid setgid,',
+ 'allow capability sys_admin,',
+ 'audit capability kill,',
+ 'deny capability chgrp, # example comment',
+ ]
+
+ for rule in rules:
+ self.ruleset.add(parse_capability(rule))
+
+ def test_ruleset_is_covered_1(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('capability chown,')))
+ def test_ruleset_is_covered_2(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('capability sys_admin,')))
+ def test_ruleset_is_covered_3(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('allow capability sys_admin,')))
+ def test_ruleset_is_covered_4(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('capability setuid,')))
+ def test_ruleset_is_covered_5(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('allow capability setgid,')))
+ def test_ruleset_is_covered_6(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('capability setgid setuid,')))
+ def test_ruleset_is_covered_7(self):
+ pass # self.assertTrue(self.ruleset.is_covered(parse_capability('capability sys_admin chown,'))) # fails because it is split over two rule objects internally
+ def test_ruleset_is_covered_8(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('capability kill,')))
+
+ def test_ruleset_is_covered_9(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability chown,')))
+ def test_ruleset_is_covered_10(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability sys_admin,')))
+ def test_ruleset_is_covered_11(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability sys_admin chown,')))
+ def test_ruleset_is_covered_12(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability setgid,')))
+ def test_ruleset_is_covered_13(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('deny capability kill,')))
+
+ def test_ruleset_is_covered_14(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability chown,')))
+ def test_ruleset_is_covered_15(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability sys_admin,')))
+ def test_ruleset_is_covered_16(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability sys_admin chown,')))
+ def test_ruleset_is_covered_17(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability setgid,')))
+ def test_ruleset_is_covered_18(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('audit capability kill,')))
+
+ def test_ruleset_is_covered_19(self):
+ self.assertTrue(self.ruleset.is_covered(parse_capability('deny capability chgrp,')))
+ def test_ruleset_is_covered_20(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('audit deny capability chgrp,')))
+ def test_ruleset_is_covered_21(self):
+ self.assertFalse(self.ruleset.is_covered(parse_capability('audit capability chgrp,')))
+
+# XXX - disabling these until we decide whether or not checking whether
+# a log is covered by rules should be a separate entry point, possibly
+# handling the log structure directly, or whether coverage should be
+# solely based on Rule objects and marshaling of a log message into a
+# Rule object should occur outside of the Rule classes themselves.
+#
+# def _test_log_covered(self, expected, capability):
+# event_base = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname="%s"'
+
+# parser = ReadLog('', '', '', '', '')
+# self.assertEqual(expected, self.ruleset.is_log_covered(parser.parse_event(event_base%capability)))
+#
+# def test_ruleset_is_log_covered_1(self):
+# self._test_log_covered(False, 'net_raw')
+# def test_ruleset_is_log_covered_2(self):
+# self._test_log_covered(True, 'chown')
+# def test_ruleset_is_log_covered_3(self):
+# self._test_log_covered(True, 'sys_admin')
+# def test_ruleset_is_log_covered_4(self):
+# self._test_log_covered(True, 'kill')
+# def test_ruleset_is_log_covered_5(self):
+# self._test_log_covered(False, 'chgrp')
+# def test_ruleset_is_log_covered_6(self):
+# event_base = 'type=AVC msg=audit(1415403814.628:662): apparmor="ALLOWED" operation="capable" profile="/bin/ping" pid=15454 comm="ping" capability=13 capname="%s"'
+#
+# parser = ReadLog('', '', '', '', '')
+# self.assertEqual(True, self.ruleset.is_log_covered(parser.parse_event(event_base%'chgrp'), False)) # ignores allow/deny
+
+class CapabilityGlobTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+ self.ruleset = CapabilityRuleset()
+
+ def test_glob(self):
+ self.assertEqual(self.ruleset.get_glob('capability net_raw,'), 'capability,')
+
+ def test_glob_ext(self):
+ with self.assertRaises(AppArmorBug):
+ self.ruleset.get_glob_ext('capability net_raw,')
+
+class CapabilityDeleteTest(unittest.TestCase):
+ def setUp(self):
+ self.maxDiff = None
+
+ self.ruleset = CapabilityRuleset()
+ rules = [
+ 'capability chown,',
+ 'allow capability sys_admin,',
+ 'deny capability chgrp, # example comment',
+ ]
+
+ for rule in rules:
+ self.ruleset.add(parse_capability(rule))
+
+ def test_delete(self):
+ expected_raw = [
+ ' capability chown,',
+ ' deny capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' capability chown,',
+ '',
+ ]
+
+ self.ruleset.delete(CapabilityRule(['sys_admin']))
+
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_with_allcaps(self):
+ expected_raw = [
+ ' capability chown,',
+ ' deny capability chgrp, # example comment',
+ ' capability,',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' capability chown,',
+ ' capability,',
+ '',
+ ]
+
+ self.ruleset.add(CapabilityRule(CapabilityRule.ALL))
+ self.ruleset.delete(CapabilityRule('sys_admin'))
+
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_with_multi(self):
+ expected_raw = [
+ ' capability chown,',
+ ' allow capability sys_admin,',
+ ' deny capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' allow capability sys_admin,',
+ ' capability chown,',
+ '',
+ ]
+
+ self.ruleset.add(CapabilityRule(['audit_read', 'audit_write']))
+ self.ruleset.delete(CapabilityRule(['audit_read', 'audit_write']))
+
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_with_multi_2(self):
+ self.ruleset.add(CapabilityRule(['audit_read', 'audit_write']))
+
+ with self.assertRaises(AppArmorBug):
+ # XXX ideally delete_raw should remove audit_read from the "capability audit_read audit_write," ruleset
+ # but that's quite some work to cover a corner case.
+ self.ruleset.delete(CapabilityRule('audit_read'))
+
+ def test_delete_raw_notfound(self):
+ with self.assertRaises(AppArmorBug):
+ self.ruleset.delete(CapabilityRule('audit_write'))
+
+ def test_delete_duplicates(self):
+ inc = CapabilityRuleset()
+ rules = [
+ 'capability chown,',
+ 'deny capability chgrp, # example comment',
+ ]
+
+ for rule in rules:
+ inc.add(parse_capability(rule))
+
+ expected_raw = [
+ ' allow capability sys_admin,',
+ '',
+ ]
+
+ expected_clean = expected_raw
+
+ self.assertEqual(self.ruleset.delete_duplicates(inc), 2)
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_duplicates_2(self):
+ inc = CapabilityRuleset()
+ rules = [
+ 'capability audit_write,',
+ 'capability chgrp, # example comment',
+ ]
+
+ for rule in rules:
+ inc.add(parse_capability(rule))
+
+ expected_raw = [
+ ' capability chown,',
+ ' allow capability sys_admin,',
+ ' deny capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' allow capability sys_admin,',
+ ' capability chown,',
+ '',
+ ]
+
+ self.assertEqual(self.ruleset.delete_duplicates(inc), 0)
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_duplicates_3(self):
+ self.ruleset.add(parse_capability('audit capability dac_override,'))
+
+ inc = CapabilityRuleset()
+ rules = [
+ 'capability dac_override,',
+ ]
+
+ for rule in rules:
+ inc.add(parse_capability(rule))
+
+ expected_raw = [
+ ' capability chown,',
+ ' allow capability sys_admin,',
+ ' deny capability chgrp, # example comment',
+ ' audit capability dac_override,',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' allow capability sys_admin,',
+ ' audit capability dac_override,',
+ ' capability chown,',
+ '',
+ ]
+
+ self.assertEqual(self.ruleset.delete_duplicates(inc), 0)
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_duplicates_4(self):
+ inc = CapabilityRuleset()
+ rules = [
+ 'capability,',
+ ]
+
+ for rule in rules:
+ inc.add(parse_capability(rule))
+
+ expected_raw = [
+ ' allow capability sys_admin,', # XXX huh? should be deleted!
+ ' deny capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' allow capability sys_admin,', # XXX huh? should be deleted!
+ '',
+ ]
+
+ self.assertEqual(self.ruleset.delete_duplicates(inc), 1)
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_duplicates_none(self):
+ expected_raw = [
+ ' capability chown,',
+ ' allow capability sys_admin,',
+ ' deny capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' allow capability sys_admin,',
+ ' capability chown,',
+ '',
+ ]
+
+ self.assertEqual(self.ruleset.delete_duplicates(None), 0)
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+ def test_delete_duplicates_hasher(self):
+ expected_raw = [
+ ' capability chown,',
+ ' allow capability sys_admin,',
+ ' deny capability chgrp, # example comment',
+ '',
+ ]
+
+ expected_clean = [
+ ' deny capability chgrp, # example comment',
+ '',
+ ' allow capability sys_admin,',
+ ' capability chown,',
+ '',
+ ]
+
+ self.assertEqual(self.ruleset.delete_duplicates(hasher()), 0)
+ self.assertEqual(expected_raw, self.ruleset.get_raw(1))
+ self.assertEqual(expected_clean, self.ruleset.get_clean(1))
+
+
+if __name__ == "__main__":
+ unittest.main(verbosity=2)
signature.asc
Description: Digital signature
-- AppArmor mailing list [email protected] Modify settings or unsubscribe at: https://lists.ubuntu.com/mailman/listinfo/apparmor
