fix: rpm extractor for windows

cve_bin_tool/extractor.py

@@ -6,10 +6,10 @@
 """
 import itertools
 import os
+import pathlib
 import re
 import shutil
 import sys
-import tarfile
 import tempfile
 
 import zstandard
@@ -106,23 +106,30 @@ async def extract_file_rpm(self, filename, extraction_path):
                 if stderr or not stdout:
                     return 1
                 filenames = await aio_glob(os.path.join(extraction_path, "*.cpio"))
+                if not filenames:
+                    filenames = await aio_glob(
+                        os.path.join(extraction_path, "*.cpio.zstd")
+                    )
+                    filename = filenames[0]
+                    exit_code = await self.extract_file_zst(filename, extraction_path)
+                    if exit_code:
+                        return 1
+                    filenames = await aio_glob(os.path.join(extraction_path, "*.cpio"))
                 filename = filenames[0]
-
                 stdout, stderr, _ = await aio_run_command(["7z", "x", filename])
                 if stderr or not stdout:
                     return 1
         return 0
 
     async def extract_file_zst(self, filename: str, extraction_path: str) -> int:
         """Extract zstd compressed files"""
-
-        dctx = zstandard.ZstdDecompressor()
+        input_file = pathlib.Path(filename)
         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()
+            with open(input_file, "rb") as compressed:
+                decomp = zstandard.ZstdDecompressor()
+                output_path = pathlib.Path(extraction_path) / input_file.stem
+                with open(output_path, "wb") as destination:
+                    decomp.copy_stream(compressed, destination)
         return e.exit_code
 
     async def extract_file_pkg(self, filename: str, extraction_path: str) -> int: