Add and use logprof_header() and logprof_header_localvars() in *Rule classes

BaseRule:
- add logprof_header() - sets the 'Qualifier' (audit, allow/deny) header
  if a qualifier is specified, calls logprof_header_localvars() and then
  returns an array of headers to display in aa-logprof and aa-mergeprof
- add logprof_header_localvars() - dummy function that needs to be
  implemented in the child classes

NetworkRule: add logprof_header_localvars() - adds 'Network Family'
and 'Socket Type' to the headers

CapabilityRule: add logprof_header_localvars() - adds 'Capability' to
the headers

Also change aa-mergeprof to use rule_obj.logprof_header() for network
and capability rules. This means deleting lots of lines (that moved to
the *Rule classes) and also deleting the last differences between
capabiltiy and network rules.

Finally add tests for the newly added functions.


Acked-by: Steve Beattie <steve@nxnw.org>

utils/aa-mergeprof

@@ -329,22 +329,7 @@ class Merge(object):
                     q.selected = default_option - 1
 
                     q.headers = [_('Profile'), apparmor.aa.combine_name(profile, hat)]
-                    qualifier = ''
+                    q.headers += rule_obj.logprof_header()
-                    if rule_obj.deny:
-                        qualifier = 'deny %s' % qualifier
-
-                    if rule_obj.audit:
-                        qualifier = 'audit %s' % qualifier
-
-                    if qualifier:
-                        q.headers += [_('Qualifier'), qualifier]
-
-                    if rule_obj.all_caps:
-                        cap_txt = 'ALL'
-                    else:
-                        cap_txt = ' '.join(rule_obj.capability)
-
-                    q.headers += [_('Capability'), cap_txt]
 
                     severity = rule_obj.severity(sev_db)
                     if severity != sev_db.NOT_IMPLEMENTED:
@@ -376,7 +361,7 @@ class Merge(object):
                             q.options = options
 
                             q.headers = [_('Profile'), apparmor.aa.combine_name(profile, hat)]
-                            q.headers += [_('Capability'), cap_txt]
+                            q.headers += rule_obj.logprof_header()
 
                         elif ans == 'CMD_ALLOW':
                             done = True
@@ -743,29 +728,7 @@ class Merge(object):
                             q.selected = default_option - 1
 
                         q.headers = [_('Profile'), apparmor.aa.combine_name(profile, hat)]
-
+                        q.headers += rule_obj.logprof_header()
-                        qualifier = ''
-                        if rule_obj.deny:
-                            qualifier = 'deny %s' % qualifier
-
-                        if rule_obj.audit:
-                            qualifier = 'audit %s' % qualifier
-
-                        if qualifier:
-                            q.headers += [_('Qualifier'), qualifier]
-
-                        if rule_obj.all_domains:
-                            family = 'ALL'
-                        else:
-                            family = rule_obj.domain
-
-                        if rule_obj.all_type_or_protocols:
-                            sock_type = 'ALL'
-                        else:
-                            sock_type = rule_obj.type_or_protocol
-
-                        q.headers += [_('Network Family'), family]
-                        q.headers += [_('Socket Type'), sock_type]
 
                         severity = rule_obj.severity(sev_db)
                         if severity != sev_db.NOT_IMPLEMENTED:
@@ -797,8 +760,7 @@ class Merge(object):
                                 q.options = options
 
                                 q.headers = [_('Profile'), apparmor.aa.combine_name(profile, hat)]
-                                q.headers += [_('Network Family'), family]
+                                q.headers += rule_obj.logprof_header()
-                                q.headers += [_('Socket Type'), sock_type]
 
                             elif ans == 'CMD_ALLOW':
                                 done = True

utils/apparmor/rule/__init__.py

@@ -143,6 +143,34 @@ class BaseRule(object):
            sev_db must be an apparmor.severity.Severity object.'''
         return sev_db.NOT_IMPLEMENTED
 
+    def logprof_header(self):
+        '''return the headers (human-readable version of the rule) to display in aa-logprof for this rule object
+           returns {'label1': 'value1', 'label2': 'value2'} '''
+
+        headers = []
+        qualifier = []
+
+        if self.audit:
+            qualifier += ['audit']
+
+        if self.deny:
+            qualifier += ['deny']
+        elif self.allow_keyword:
+            qualifier += ['allow']
+
+        if qualifier:
+            headers += [_('Qualifier'), ' '.join(qualifier)]
+
+        headers += self.logprof_header_localvars()
+
+        return headers
+
+    # @abstractmethod  FIXME - uncomment when python3 only
+    def logprof_header_localvars(self):
+        '''return the headers (human-readable version of the rule) to display in aa-logprof for this rule object
+           returns {'label1': 'value1', 'label2': 'value2'} '''
+        raise AppArmorBug("'%s' needs to implement logprof_header(), but didn't" % (str(self)))
+
     def modifiers_str(self):
         '''return the allow/deny and audit keyword as string, including whitespace'''
 

utils/apparmor/rule/capability.py

@@ -141,6 +141,17 @@ class CapabilityRule(BaseRule):
 
         return severity
 
+    def logprof_header_localvars(self):
+        if self.all_caps:
+            cap_txt = _('ALL')
+        else:
+            cap_txt = ' '.join(sorted(self.capability))
+
+        return [
+            _('Capability'), cap_txt,
+        ]
+
+
 class CapabilityRuleset(BaseRuleset):
     '''Class to handle and store a collection of capability rules'''
 

utils/apparmor/rule/network.py

@@ -199,6 +199,22 @@ class NetworkRule(BaseRule):
 
         return True
 
+    def logprof_header_localvars(self):
+        if self.all_domains:
+            family = _('ALL')
+        else:
+            family = self.domain
+
+        if self.all_type_or_protocols:
+            sock_type = _('ALL')
+        else:
+            sock_type = self.type_or_protocol
+
+        return [
+            _('Network Family'), family,
+            _('Socket Type'),    sock_type,
+        ]
+
 
 class NetworkRuleset(BaseRuleset):
     '''Class to handle and store a collection of network rules'''

utils/test/test-baserule.py

@@ -58,6 +58,11 @@ class TestBaserule(AATest):
         rank = obj.severity(sev_db)
         self.assertEqual(rank, sev_db.NOT_IMPLEMENTED)
 
+    def test_logprof_header_localvars(self):
+        obj = BaseRule()
+        with self.assertRaises(AppArmorBug):
+            obj.logprof_header_localvars()
+
 
 setup_all_loops(__name__)
 if __name__ == '__main__':

utils/test/test-capability.py

@@ -21,6 +21,8 @@ from apparmor.rule import BaseRule
 import apparmor.severity as severity
 from apparmor.common import AppArmorException, AppArmorBug, hasher
 from apparmor.logparser import ReadLog
+from apparmor.translations import init_translation
+_ = init_translation()
 
 # --- tests for single CapabilityRule --- #
 
@@ -434,6 +436,21 @@ class CapabiliySeverityTest(AATest):
         rank = obj.severity(sev_db)
         self.assertEqual(rank, expected)
 
+class CapabilityLogprofHeaderTest(AATest):
+    tests = [
+        ('capability,',                         [                               _('Capability'), _('ALL'),         ]),
+        ('capability chown,',                   [                               _('Capability'), 'chown',          ]),
+        ('capability chown fsetid,',            [                               _('Capability'), 'chown fsetid',   ]),
+        ('audit capability,',                   [_('Qualifier'), 'audit',       _('Capability'), _('ALL'),         ]),
+        ('deny capability chown,',              [_('Qualifier'), 'deny',        _('Capability'), 'chown',          ]),
+        ('allow capability chown fsetid,',      [_('Qualifier'), 'allow',       _('Capability'), 'chown fsetid',   ]),
+        ('audit deny capability,',              [_('Qualifier'), 'audit deny',  _('Capability'), _('ALL'),         ]),
+    ]
+
+    def _run_test(self, params, expected):
+        obj = CapabilityRule._parse(params)
+        self.assertEqual(obj.logprof_header(), expected)
+
 # --- tests for CapabilityRuleset --- #
 
 class CapabilityRulesTest(AATest):

utils/test/test-network.py

@@ -21,6 +21,8 @@ from apparmor.rule.network import NetworkRule, NetworkRuleset
 from apparmor.rule import BaseRule
 from apparmor.common import AppArmorException, AppArmorBug
 from apparmor.logparser import ReadLog
+from apparmor.translations import init_translation
+_ = init_translation()
 
 exp = namedtuple('exp', ['audit', 'allow_keyword', 'deny', 'comment',
         'domain', 'all_domains', 'type_or_protocol', 'all_type_or_protocols'])
@@ -336,6 +338,21 @@ class NetworkCoveredTest_Invalid(AATest):
         with self.assertRaises(AppArmorBug):
             obj.is_equal(testobj)
 
+class NetworkLogprofHeaderTest(AATest):
+    tests = [
+        ('network,',                    [                               _('Network Family'), _('ALL'),     _('Socket Type'), _('ALL'),     ]),
+        ('network inet,',               [                               _('Network Family'), 'inet',       _('Socket Type'), _('ALL'),     ]),
+        ('network inet stream,',        [                               _('Network Family'), 'inet',       _('Socket Type'), 'stream',     ]),
+        ('deny network,',               [_('Qualifier'), 'deny',        _('Network Family'), _('ALL'),     _('Socket Type'), _('ALL'),     ]),
+        ('allow network inet,',         [_('Qualifier'), 'allow',       _('Network Family'), 'inet',       _('Socket Type'), _('ALL'),     ]),
+        ('audit network inet stream,',  [_('Qualifier'), 'audit',       _('Network Family'), 'inet',       _('Socket Type'), 'stream',     ]),
+        ('audit deny network inet,',    [_('Qualifier'), 'audit deny',  _('Network Family'), 'inet',       _('Socket Type'), _('ALL'),     ]),
+    ]
+
+    def _run_test(self, params, expected):
+        obj = NetworkRule._parse(params)
+        self.assertEqual(obj.logprof_header(), expected)
+
 ## --- tests for NetworkRuleset --- #
 
 class NetworkRulesTest(AATest):