feat: extractor support for .pkg and .zst packages (#1580)

Author: yashugarg

cve_bin_tool/extractor.py

@@ -9,8 +9,10 @@
 import re
 import shutil
 import sys
+import tarfile
 import tempfile
 
+import zstandard
 from rpmfile.cli import main as rpmextract
 
 from cve_bin_tool.async_utils import (
@@ -47,6 +49,8 @@ def __init__(self, logger=None, error_mode=ErrorMode.TruncTrace):
             self.extract_file_deb: {".deb", ".ipk"},
             self.extract_file_cab: {".cab"},
             self.extract_file_apk: {".apk"},
+            self.extract_file_zst: {".zst"},
+            self.extract_file_pkg: {".pkg"},
             self.extract_file_zip: {
                 ".exe",
                 ".zip",
@@ -109,6 +113,40 @@ async def extract_file_rpm(self, filename, extraction_path):
                     return 1
         return 0
 
+    async def extract_file_zst(self, filename: str, extraction_path: str) -> int:
+        """Extract zstd compressed files"""
+
+        dctx = zstandard.ZstdDecompressor()
+        with ErrorHandler(mode=ErrorMode.Ignore) as e:
+            with open(filename, "rb") as f:
+                with dctx.stream_reader(f) as reader:
+                    with tarfile.open(mode="r|", fileobj=reader) as tar:
+                        tar.extractall(extraction_path)
+                    tar.close()
+        return e.exit_code
+
+    async def extract_file_pkg(self, filename: str, extraction_path: str) -> int:
+        """Extract pkg files"""
+
+        # Tarfile wasn't used here because it can't open [.pkg] files directy
+        # and failed to manage distinct compression types in differnet versions of FreeBSD packages.
+        # Reference: https://github.com/intel/cve-bin-tool/pull/1580#discussion_r829346602
+        if await aio_inpath("tar"):
+            stdout, stderr, return_code = await aio_run_command(
+                ["tar", "xf", filename, "-C", extraction_path]
+            )
+            if (stderr or not stdout) and return_code != 0:
+                return 1
+            return 0
+        if await aio_inpath("7z"):
+            stdout, stderr, _ = await aio_run_command(
+                ["7z", "x", filename, "-o", extraction_path]
+            )
+            if stderr or not stdout:
+                return 1
+            return 0
+        return 1
+
     async def extract_file_deb(self, filename, extraction_path):
         """Extract debian packages"""
         if not await aio_inpath("ar"):