fix: extract apk packages for alpine and android (#1258)

Author: imsahil007

cve_bin_tool/async_utils.py

@@ -73,18 +73,18 @@ def run_coroutine(coro):
     return result
 
 
-async def aio_run_command(args):
+async def aio_run_command(args, process_can_fail=False):
     process = await asyncio.create_subprocess_exec(
         *args,
         stdout=asyncio.subprocess.PIPE,
         stderr=asyncio.subprocess.PIPE,
     )
     stdout, stderr = await process.communicate()
-    if process.returncode != 0:
+    if process.returncode != 0 and not process_can_fail:
         raise subprocess.CalledProcessError(
             args, process.returncode, output=stdout, stderr=stderr
         )
-    return stdout, stderr  # binary encoded
+    return stdout, stderr, process.returncode  # binary encoded
 
 
 class ChangeDirContext:

cve_bin_tool/extractor.py

@@ -8,6 +8,7 @@
 """
 import itertools
 import os
+import re
 import shutil
 import sys
 import tempfile
@@ -47,11 +48,11 @@ def __init__(self, logger=None, error_mode=ErrorMode.TruncTrace):
             self.extract_file_rpm: {".rpm"},
             self.extract_file_deb: {".deb", ".ipk"},
             self.extract_file_cab: {".cab"},
+            self.extract_file_apk: {".apk"},
             self.extract_file_zip: {
                 ".exe",
                 ".zip",
                 ".jar",
-                ".apk",
                 ".msi",
                 ".egg",
                 ".whl",
@@ -78,13 +79,13 @@ async def extract_file_rpm(self, filename, extraction_path):
             if not await aio_inpath("rpm2cpio") or not await aio_inpath("cpio"):
                 await rpmextract("-xC", extraction_path, filename)
             else:
-                stdout, stderr = await aio_run_command(["rpm2cpio", filename])
+                stdout, stderr, _ = await aio_run_command(["rpm2cpio", filename])
                 if stderr or not stdout:
                     return 1
                 cpio_path = os.path.join(extraction_path, "data.cpio")
                 async with FileIO(cpio_path, "wb") as f:
                     await f.write(stdout)
-                stdout, stderr = await aio_run_command(
+                stdout, stderr, _ = await aio_run_command(
                     ["cpio", "-idm", "--file", cpio_path]
                 )
                 if stdout or not stderr:
@@ -94,13 +95,13 @@ async def extract_file_rpm(self, filename, extraction_path):
                 with ErrorHandler(mode=self.error_mode, logger=self.logger):
                     raise Exception("7z is required to extract rpm files")
             else:
-                stdout, stderr = await aio_run_command(["7z", "x", filename])
+                stdout, stderr, _ = await aio_run_command(["7z", "x", filename])
                 if stderr or not stdout:
                     return 1
                 filenames = await aio_glob(os.path.join(extraction_path, "*.cpio"))
                 filename = filenames[0]
 
-                stdout, stderr = await aio_run_command(["7z", "x", filename])
+                stdout, stderr, _ = await aio_run_command(["7z", "x", filename])
                 if stderr or not stdout:
                     return 1
         return 0
@@ -111,45 +112,82 @@ async def extract_file_deb(self, filename, extraction_path):
             with ErrorHandler(mode=self.error_mode, logger=self.logger):
                 raise Exception("'ar' is required to extract deb files")
         else:
-            stdout, stderr = await aio_run_command(["ar", "x", filename])
+            stdout, stderr, _ = await aio_run_command(["ar", "x", filename])
             if stderr:
                 return 1
             datafile = await aio_glob(os.path.join(extraction_path, "data.tar.*"))
             with ErrorHandler(mode=ErrorMode.Ignore) as e:
                 await aio_unpack_archive(datafile[0], extraction_path)
             return e.exit_code
 
+    async def extract_file_apk(self, filename, extraction_path):
+        """Check whether it is alpine or android package"""
+
+        is_tar = True
+        process_can_fail = True
+        if await aio_inpath("unzip"):
+            stdout, stderr, return_code = await aio_run_command(
+                ["unzip", "-l", filename], process_can_fail
+            )
+            if return_code == 0:
+                is_tar = False
+        elif await aio_inpath("7z"):
+            stdout, stderr, return_code = await aio_run_command(
+                ["7z", "t", filename], process_can_fail
+            )
+            if re.search(b"Type = Zip", stdout):
+                is_tar = False
+        elif await aio_inpath("zipinfo"):
+            stdout, stderr, return_code = await aio_run_command(
+                ["zipinfo", filename], process_can_fail
+            )
+            if return_code == 0:
+                is_tar = False
+        elif await aio_inpath("file"):
+            stdout, stderr, return_code = await aio_run_command(
+                ["file", filename], process_can_fail
+            )
+            if re.search(b"Zip archive data", stdout):
+                is_tar = False
+        if is_tar:
+            self.logger.debug(f"Extracting {filename} as a tar.gzip file")
+            with ErrorHandler(mode=ErrorMode.Ignore) as e:
+                await aio_unpack_archive(filename, extraction_path, format="gztar")
+            return e.exit_code
+        else:
+            return await self.extract_file_zip(filename, extraction_path)
+
     @staticmethod
     async def extract_file_cab(filename, extraction_path):
         """Extract cab files"""
         if sys.platform.startswith("linux"):
             if not await aio_inpath("cabextract"):
                 raise Exception("'cabextract' is required to extract cab files")
             else:
-                stdout, stderr = await aio_run_command(
+                stdout, stderr, _ = await aio_run_command(
                     ["cabextract", "-d", extraction_path, filename]
                 )
                 if stderr or not stdout:
                     return 1
         else:
-            stdout, stderr = await aio_run_command(
+            stdout, stderr, _ = await aio_run_command(
                 ["Expand", filename, "-R -F:*", extraction_path]
             )
             if stderr or not stdout:
                 return 1
         return 0
 
     @staticmethod
-    async def extract_file_zip(filename, extraction_path):
+    async def extract_file_zip(filename, extraction_path, process_can_fail=False):
         """Extract zip files"""
         if await aio_inpath("unzip"):
-            stdout, stderr = await aio_run_command(
+            stdout, stderr, _ = await aio_run_command(
                 ["unzip", "-n", "-d", extraction_path, filename]
             )
             if stderr or not stdout:
                 return 1
         elif await aio_inpath("7z"):
-            stdout, stderr = await aio_run_command(["7z", "x", filename])
+            stdout, stderr, _ = await aio_run_command(["7z", "x", filename])
             if stderr or not stdout:
                 return 1
         else:

test/test_strings.py

@@ -27,7 +27,9 @@ async def _parse_test(self, filename):
         """Helper function to parse a binary file and check whether
         the given string is in the parsed result"""
         self.strings.filename = os.path.join(ASSETS_PATH, filename)
-        binutils_strings, _ = await aio_run_command(["strings", self.strings.filename])
+        binutils_strings, _, _ = await aio_run_command(
+            ["strings", self.strings.filename]
+        )
         ours = await self.strings.aio_parse()
         for theirs in binutils_strings.splitlines():
             assert theirs.decode("utf-8") in ours