unarchive: Fail when zipinfo is not available (#74632)

Fixes: #39029

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

changelogs/fragments/39029_unarchive.yml

@@ -0,0 +1,2 @@
+bugfixes:
+- unarchive - fail when zipinfo binary is not found in executable paths (https://github.com/ansible/ansible/issues/39029).

lib/ansible/modules/unarchive.py

@@ -238,6 +238,7 @@ from zipfile import ZipFile, BadZipfile
 from ansible.module_utils.basic import AnsibleModule
 from ansible.module_utils.urls import fetch_file
 from ansible.module_utils._text import to_bytes, to_native, to_text
+from ansible.module_utils.common.process import get_bin_path
 
 try:  # python 3.3+
     from shlex import quote
@@ -289,8 +290,14 @@ class ZipArchive(object):
         self.excludes = module.params['exclude']
         self.includes = []
         self.include_files = self.module.params['include']
-        self.cmd_path = self.module.get_bin_path('unzip')
+        try:
-        self.zipinfocmd_path = self.module.get_bin_path('zipinfo')
+            self.cmd_path = get_bin_path('unzip')
+        except ValueError:
+            self.module.fail_json(msg="Unable to find required 'unzip' binary in the path")
+        try:
+            self.zipinfocmd_path = get_bin_path('zipinfo')
+        except ValueError:
+            self.module.fail_json(msg="Unable to find required 'zipinfo' binary in the path")
         self._files_in_archive = []
         self._infodict = dict()
 
@@ -308,11 +315,7 @@ class ZipArchive(object):
         return (mode & ~umask)
 
     def _legacy_file_list(self):
-        unzip_bin = self.module.get_bin_path('unzip')
+        rc, out, err = self.module.run_command([self.cmd_path, '-v', self.src])
-        if not unzip_bin:
-            raise UnarchiveError('Python Zipfile cannot read %s and unzip not found' % self.src)
-
-        rc, out, err = self.module.run_command([unzip_bin, '-v', self.src])
         if rc:
             raise UnarchiveError('Neither python zipfile nor unzip can read %s' % self.src)
 
@@ -385,6 +388,7 @@ class ZipArchive(object):
     def is_unarchived(self):
         # BSD unzip doesn't support zipinfo listings with timestamp.
         cmd = [self.zipinfocmd_path, '-T', '-s', self.src]
+
         if self.excludes:
             cmd.extend(['-x', ] + self.excludes)
         if self.include_files:
@@ -726,10 +730,14 @@ class TgzArchive(object):
         self.excludes = [path.rstrip('/') for path in self.module.params['exclude']]
         self.include_files = self.module.params['include']
         # Prefer gtar (GNU tar) as it supports the compression options -z, -j and -J
-        self.cmd_path = self.module.get_bin_path('gtar', None)
+        try:
-        if not self.cmd_path:
+            self.cmd_path = get_bin_path('gtar')
+        except ValueError:
             # Fallback to tar
-            self.cmd_path = self.module.get_bin_path('tar')
+            try:
+                self.cmd_path = get_bin_path('tar')
+            except ValueError:
+                self.module.fail_json(msg="Unable to find required 'gtar' or 'tar' binary in the path")
         self.zipflag = '-z'
         self._files_in_archive = []
 
@@ -865,9 +873,6 @@ class TgzArchive(object):
         return dict(cmd=cmd, rc=rc, out=out, err=err)
 
     def can_handle_archive(self):
-        if not self.cmd_path:
-            return False, 'Commands "gtar" and "tar" not found.'
-
         if self.tar_type != 'gnu':
             return False, 'Command "%s" detected as tar type %s. GNU tar required.' % (self.cmd_path, self.tar_type)
 

test/units/modules/test_unarchive.py

@@ -0,0 +1,76 @@
+from __future__ import absolute_import, division, print_function
+
+__metaclass__ = type
+
+
+import pytest
+
+from ansible.modules.unarchive import ZipArchive, TgzArchive
+
+
+class AnsibleModuleExit(Exception):
+    def __init__(self, *args, **kwargs):
+        self.args = args
+        self.kwargs = kwargs
+
+
+class ExitJson(AnsibleModuleExit):
+    pass
+
+
+class FailJson(AnsibleModuleExit):
+    pass
+
+
+@pytest.fixture
+def fake_ansible_module():
+    return FakeAnsibleModule()
+
+
+class FakeAnsibleModule:
+    def __init__(self):
+        self.params = {}
+        self.tmpdir = None
+
+    def exit_json(self, *args, **kwargs):
+        raise ExitJson(*args, **kwargs)
+
+    def fail_json(self, *args, **kwargs):
+        raise FailJson(*args, **kwargs)
+
+
+class TestCaseZipArchive:
+    def test_no_zip_binary(self, mocker, fake_ansible_module):
+        mocker.patch("ansible.modules.unarchive.get_bin_path", side_effect=ValueError)
+        fake_ansible_module.params = {
+            "extra_opts": "",
+            "exclude": "",
+            "include": "",
+        }
+        with pytest.raises(FailJson) as exec_info:
+            z = ZipArchive(
+                src="",
+                b_dest="",
+                file_args="",
+                module=fake_ansible_module,
+            )
+        assert "Unable to find required" in exec_info.value.kwargs["msg"]
+
+
+class TestCaseTgzArchive:
+    def test_no_tar_binary(self, mocker, fake_ansible_module):
+        mocker.patch("ansible.modules.unarchive.get_bin_path", side_effect=ValueError)
+        fake_ansible_module.params = {
+            "extra_opts": "",
+            "exclude": "",
+            "include": "",
+        }
+        fake_ansible_module.check_mode = False
+        with pytest.raises(FailJson) as exec_info:
+            z = TgzArchive(
+                src="",
+                b_dest="",
+                file_args="",
+                module=fake_ansible_module,
+            )
+        assert "Unable to find required" in exec_info.value.kwargs["msg"]