From 588d00775470be40bec6bf2cad2f819d144fb3d1 Mon Sep 17 00:00:00 2001
From: Kim Mason <kim.mason@datadoghq.com>
Date: Tue, 25 Mar 2025 00:40:53 +0000
Subject: [PATCH] Make extraction work for MSI files when unzip and 7zip
 installed.

This commit fixes an issue where if both unzip and 7zip were installed,
MSI file extraction failed because unzip was attempted first, and failed.

Now in the case where both are installed, unzip is tried, and if that
fails, then 7zip is tried.
---
 cve_bin_tool/extractor.py | 75 +++++++++++++++++++++++----------------
 1 file changed, 45 insertions(+), 30 deletions(-)

diff --git a/cve_bin_tool/extractor.py b/cve_bin_tool/extractor.py
index 8f4e3017a3..7e2b79b9d8 100644
--- a/cve_bin_tool/extractor.py
+++ b/cve_bin_tool/extractor.py
@@ -46,6 +46,45 @@
 EXTENSIONS = "extensions"
 MIMES = "mimes"
 
+# Invalid key for unzip and 7z extraction to prevent freezing on password
+# protected files.
+STATIC_INVALID_KEY = "StaticInvalidKey"
+
+
+async def unzip_file(filename, extraction_path, process_can_fail):
+    """Extracts files using the unzip utility."""
+    stdout, stderr, _ = await aio_run_command(
+        ["unzip", "-P", STATIC_INVALID_KEY, "-n", "-d", extraction_path, filename],
+        process_can_fail,
+    )
+    if not stderr:
+        return 0
+    if "incorrect password" in stderr.decode():
+        LOGGER.error(f"Failed to extract {filename}: The file is password protected")
+        return 0
+    is_exe = filename.endswith(".exe")
+    if is_exe:
+        return 0  # not all .exe files are zipfiles, no need for error
+    return 1
+
+
+async def unzip_7z(filename, extraction_path, process_can_fail):
+    """Extracts files using the 7z utility. 7z supports more file format than\
+        unzip does.
+    """
+    stdout, stderr, _ = await aio_run_command(
+        ["7z", "x", f"-p{STATIC_INVALID_KEY}", filename], process_can_fail
+    )
+    if stdout and not stderr:
+        return 0
+    if "Wrong password" in stderr.decode():
+        LOGGER.error(f"Failed to extract {filename}: The file is password protected")
+        return 0
+    is_exe = filename.endswith(".exe")
+    if is_exe:
+        return 0  # not all .exe files are zipfiles, no need for error
+    return 1
+
 
 class BaseExtractor:
     """Extracts tar, rpm, etc. files"""
@@ -382,41 +421,17 @@ async def extract_file_zip(filename, extraction_path, process_can_fail=True):
         freezing during extraction if they are password protected.
         Providing a key during extraction has no effect if the zip file is
         not password protected and extraction will happen as normal."""
-
-        is_exe = filename.endswith(".exe")
-        key = "StaticInvalidKey"
         if await aio_inpath("unzip"):
-            stdout, stderr, _ = await aio_run_command(
-                ["unzip", "-P", key, "-n", "-d", extraction_path, filename],
-                process_can_fail,
-            )
-            if stderr:
-                if "incorrect password" in stderr.decode():
-                    LOGGER.error(
-                        f"Failed to extract {filename}: The file is password protected"
-                    )
-                    return 0
-                if is_exe:
-                    return 0  # not all .exe files are zipfiles, no need for error
-                return 1
-        elif await aio_inpath("7z"):
-            stdout, stderr, _ = await aio_run_command(
-                ["7z", "x", f"-p{key}", filename], process_can_fail
-            )
-            if stderr or not stdout:
-                if "Wrong password" in stderr.decode():
-                    LOGGER.error(
-                        f"Failed to extract {filename}: The file is password protected"
-                    )
-                    return 0
-                if is_exe:
-                    return 0  # not all .exe files are zipfiles, no need for error
-                return 1
+            result = await unzip_file(filename, extraction_path, process_can_fail)
+            if result == 0:
+                return result
+            LOGGER.debug(f"Failed to extract {filename} using unzip. Trying 7z.")
+        if await aio_inpath("7z"):
+            return await unzip_7z(filename, extraction_path, process_can_fail)
         else:
             with ErrorHandler(mode=ErrorMode.Ignore) as e:
                 await aio_unpack_archive(filename, extraction_path)
             return e.exit_code
-        return 0
 
 
 class TempDirExtractorContext(BaseExtractor):