test: tests for capsh

Signed-off-by: Abhijeet Kasurde <akasurde@redhat.com>

lib/ansible/module_utils/facts/system/caps.py

@@ -1,19 +1,6 @@
+# Copyright: Contributors to the Ansible project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
 # Collect facts related to systems 'capabilities' via capsh
-#
-# This file is part of Ansible
-#
-# Ansible is free software: you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation, either version 3 of the License, or
-# (at your option) any later version.
-#
-# Ansible 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.
-#
-# You should have received a copy of the GNU General Public License
-# along with Ansible.  If not, see <http://www.gnu.org/licenses/>.
 
 from __future__ import annotations
 
@@ -27,34 +14,55 @@ class SystemCapabilitiesFactCollector(BaseFactCollector):
     _fact_ids = set(['system_capabilities',
                      'system_capabilities_enforced'])  # type: t.Set[str]
 
+    def get_caps_data(self, module=None):
+        rc, out, err = (-1, '', '')
+        capsh_path = module.get_bin_path('capsh')
+        if capsh_path is None:
+            return rc, out, err
+
+        try:
+            rc, out, err = module.run_command(
+                [capsh_path, "--print"],
+                errors='surrogate_then_replace',
+                handle_exceptions=False
+            )
+        except (IOError, OSError) as e:
+            module.warn('Could not query system capabilities: %s' % str(e))
+
+        return rc, out, err
+
+    def parse_caps_data(self, caps_data=None):
+        enforced = 'NA'
+        enforced_caps = []
+        if caps_data is None:
+            return enforced, enforced_caps
+
+        for line in caps_data.splitlines():
+            if len(line) < 1:
+                continue
+            if line.startswith('Current:'):
+                if line.split(':')[1].strip() == '=ep':
+                    enforced = 'False'
+                else:
+                    enforced = 'True'
+                    enforced_caps = [i.strip() for i in line.split('=')[1].split(',')]
+
+        return enforced, enforced_caps
+
     def collect(self, module=None, collected_facts=None):
+        facts_dict = {
+            'system_capabilities_enforced': 'N/A',
+            'system_capabilities': 'N/A'
+        }
+        if module is None:
+            return facts_dict
+
+        rc, out, dummy = self.get_caps_data(module=module)
 
-        rc = -1
+        if rc != 0:
-        facts_dict = {'system_capabilities_enforced': 'N/A',
+            return facts_dict
-                      'system_capabilities': 'N/A'}
-        if module:
-            capsh_path = module.get_bin_path('capsh')
-            if capsh_path:
-                # NOTE: -> get_caps_data()/parse_caps_data() for easier mocking -akl
-                try:
-                    rc, out, err = module.run_command([capsh_path, "--print"], errors='surrogate_then_replace', handle_exceptions=False)
-                except (IOError, OSError) as e:
-                    module.warn('Could not query system capabilities: %s' % str(e))
-
-            if rc == 0:
-                enforced_caps = []
-                enforced = 'NA'
-                for line in out.splitlines():
-                    if len(line) < 1:
-                        continue
-                    if line.startswith('Current:'):
-                        if line.split(':')[1].strip() == '=ep':
-                            enforced = 'False'
-                        else:
-                            enforced = 'True'
-                            enforced_caps = [i.strip() for i in line.split('=')[1].split(',')]
-
-                facts_dict['system_capabilities_enforced'] = enforced
-                facts_dict['system_capabilities'] = enforced_caps
 
+        enforced, enforced_caps = self.parse_caps_data(caps_data=out)
+        facts_dict['system_capabilities_enforced'] = enforced
+        facts_dict['system_capabilities'] = enforced_caps
         return facts_dict

test/units/module_utils/facts/system/fixtures/capsh_hybrid.txt

@@ -0,0 +1,9 @@
+Current: = cap_chown,cap_dac_override,cap_fowner,cap_fsetid,cap_kill,cap_setgid,cap_setuid,cap_setpcap,cap_net_bind_service,cap_net_raw,cap_sys_chroot,cap_mknod,cap_audit_write,cap_setfcap+eip
+Bounding set =cap_chown,cap_dac_override,cap_fowner,cap_fsetid,cap_kill,cap_setgid,cap_setuid,cap_setpcap,cap_net_bind_service,cap_net_raw,cap_sys_chroot,cap_mknod,cap_audit_write,cap_setfcap
+Securebits: 00/0x0/1'b0
+secure-noroot: no (unlocked)
+secure-no-suid-fixup: no (unlocked)
+secure-keep-caps: no (unlocked)
+uid=0(root)
+gid=0(root)
+groups=

test/units/module_utils/facts/system/fixtures/capsh_uncertain.txt

@@ -0,0 +1,13 @@
+Current: =ep
+Bounding set =cap_chown,cap_dac_override,cap_dac_read_search,cap_fowner,cap_fsetid,cap_kill,cap_setgid,cap_setuid,cap_setpcap,cap_linux_immutable,cap_net_bind_service,cap_net_broadcast,cap_net_admin,cap_net_raw,cap_ipc_lock,cap_ipc_owner,cap_sys_module,cap_sys_rawio,cap_sys_chroot,cap_sys_ptrace,cap_sys_pacct,cap_sys_admin,cap_sys_boot,cap_sys_nice,cap_sys_resource,cap_sys_time,cap_sys_tty_config,cap_mknod,cap_lease,cap_audit_write,cap_audit_control,cap_setfcap,cap_mac_override,cap_mac_admin,cap_syslog,cap_wake_alarm,cap_block_suspend,cap_audit_read,cap_perfmon,cap_bpf,cap_checkpoint_restore
+Ambient set =
+Current IAB:
+Securebits: 00/0x0/1'b0 (no-new-privs=0)
+ secure-noroot: no (unlocked)
+ secure-no-suid-fixup: no (unlocked)
+ secure-keep-caps: no (unlocked)
+ secure-no-ambient-raise: no (unlocked)
+uid=0(root) euid=0(root)
+gid=0(root)
+groups=0(root)
+Guessed mode: UNCERTAIN (0)

test/units/module_utils/facts/system/test_system_caps.py

@@ -0,0 +1,55 @@
+# Copyright: Contributors to the Ansible project
+# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
+
+from __future__ import annotations
+
+import pathlib
+
+from ansible.module_utils.facts.system.caps import SystemCapabilitiesFactCollector
+
+
+class TestSystemCapsFacts:
+    fixtures = pathlib.Path(__file__).parent / "fixtures"
+
+    def _get_mock_capsh_data(self, *args, **kwargs):
+        return
+
+    def test_capsh_collect_no_data(self):
+        cap_mgr = SystemCapabilitiesFactCollector().collect()
+        assert "system_capabilities_enforced" in cap_mgr
+        assert "system_capabilities" in cap_mgr
+        assert cap_mgr["system_capabilities"] == "N/A"
+        assert cap_mgr["system_capabilities_enforced"] == "N/A"
+
+    def test_capsh_collect_uncertain(self, mocker):
+        module = mocker.MagicMock()
+        mocked_output = (self.fixtures / "capsh_uncertain.txt").read_text()
+        cap_mgr = SystemCapabilitiesFactCollector().parse_caps_data(
+            caps_data=mocked_output
+        )
+        assert cap_mgr[0] == "False"
+        assert cap_mgr[1] == []
+
+    def test_capsh_collect_hybrid(self, mocker):
+        module = mocker.MagicMock()
+        mocked_output = (self.fixtures / "capsh_hybrid.txt").read_text()
+        cap_mgr = SystemCapabilitiesFactCollector().parse_caps_data(
+            caps_data=mocked_output
+        )
+        assert cap_mgr[0] == "True"
+        assert cap_mgr[1] == [
+            "cap_chown",
+            "cap_dac_override",
+            "cap_fowner",
+            "cap_fsetid",
+            "cap_kill",
+            "cap_setgid",
+            "cap_setuid",
+            "cap_setpcap",
+            "cap_net_bind_service",
+            "cap_net_raw",
+            "cap_sys_chroot",
+            "cap_mknod",
+            "cap_audit_write",
+            "cap_setfcap+eip",
+        ]