refactor: use pathlib.Path instead of os.path (#1714)

* refactor: use pathlib.Path instead of os.path

* refactor(test_extractor): use pathlib.Path instead of os.path

* refactor(test_requirements): use pathlib.Path instead of os.path

* refactor(test_input_engine): use pathlib.Path instead of os.path

* refactor(test_json): use pathlib.Path instead of os.path

* refactor(test_file): use pathlib.Path instead of os.path

* refactor(test_config): use pathlib.Path instead of os.path

* refactor(test_merge): use pathlib.Path instead of os.path

* refactor(test_strings): use pathlib.Path instead of os.path

* refactor(test_language_scanner): use pathlib.Path instead of os.path

* refactor(test_package_list_parser): use pathlib.Path instead of os.path

* refactor(test_output_engine): use pathlib.Path instead of os.path

* refactor(test_sbom): use pathlib.Path instead of os.path

* refactor(test_cli): use pathlib.Path instead of os.path

* refactor(test_scanner): use pathlib.Path instead of os.path

* fix: unsupported operand

Author: b31ngd3v

cve_bin_tool/extractor.py

@@ -5,12 +5,12 @@
 Extraction of archives
 """
 import itertools
-import os
 import re
 import shutil
 import sys
 import tarfile
 import tempfile
+from pathlib import Path
 
 import zstandard
 from rpmfile.cli import main as rpmextract
@@ -66,7 +66,7 @@ def __init__(self, logger=None, error_mode=ErrorMode.TruncTrace):
     def can_extract(self, filename):
         """Check if the filename is something we know how to extract"""
         # Do not try to extract symlinks
-        if os.path.islink(filename):
+        if Path(filename).is_symlink():
             return False
         for extension in itertools.chain(*self.file_extractors.values()):
             if filename.endswith(extension):
@@ -82,14 +82,15 @@ async def extract_file_tar(filename, extraction_path):
 
     async def extract_file_rpm(self, filename, extraction_path):
         """Extract rpm packages"""
+        extraction_path_pathlib = Path(extraction_path)
         if sys.platform.startswith("linux"):
             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])
                 if stderr or not stdout:
                     return 1
-                cpio_path = os.path.join(extraction_path, "data.cpio")
+                cpio_path = str(extraction_path_pathlib / "data.cpio")
                 async with FileIO(cpio_path, "wb") as f:
                     await f.write(stdout)
                 stdout, stderr, _ = await aio_run_command(
@@ -105,16 +106,16 @@ async def extract_file_rpm(self, filename, extraction_path):
                 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"))
+                filenames = await aio_glob(str(extraction_path_pathlib / "*.cpio"))
                 if not filenames:
                     filenames = await aio_glob(
-                        os.path.join(extraction_path, "*.cpio.zstd")
+                        str(extraction_path_pathlib / "*.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"))
+                    filenames = await aio_glob(str(extraction_path_pathlib / "*.cpio"))
                 filename = filenames[0]
                 stdout, stderr, _ = await aio_run_command(["7z", "x", filename])
                 if stderr or not stdout:
@@ -128,9 +129,7 @@ async def extract_file_zst(self, filename: str, extraction_path: str) -> int:
         with ErrorHandler(mode=ErrorMode.Ignore) as e:
             if filename.endswith(".cpio.zstd"):
                 with open(filename, "rb") as compressed:
-                    base = os.path.basename(filename)
-                    file = os.path.splitext(base)[0]
-                    output_path = os.path.join(extraction_path, file)
+                    output_path = Path(extraction_path) / Path(filename).stem
                     with open(output_path, "wb") as destination:
                         dctx.copy_stream(compressed, destination)
             else:
@@ -147,12 +146,12 @@ async def extract_file_pkg(self, filename: str, extraction_path: str) -> int:
         async def _extract_through_7z() -> int:
             """Extract file using `7z`"""
 
-            temp = os.path.join(self.tempdir, f"{os.path.basename(filename)}")
+            temp = str(Path(self.tempdir) / Path(filename).stem)
             stdout, stderr, _ = await aio_run_command(
                 ["7z", "x", filename, f"-o{self.tempdir}"]
             )
             stdout, stderr, _ = await aio_run_command(
-                ["7z", "x", temp[:-4], f"-o{extraction_path}"]
+                ["7z", "x", temp, f"-o{extraction_path}"]
             )
             if not stdout:
                 return 1
@@ -185,7 +184,7 @@ async def extract_file_deb(self, filename, extraction_path):
             stdout, stderr, _ = await aio_run_command(["ar", "x", filename])
             if stderr:
                 return 1
-            datafile = await aio_glob(os.path.join(extraction_path, "data.tar.*"))
+            datafile = await aio_glob(str(Path(extraction_path) / "data.tar.*"))
             with ErrorHandler(mode=ErrorMode.Ignore) as e:
                 await aio_unpack_archive(datafile[0], extraction_path)
             return e.exit_code
@@ -285,15 +284,16 @@ def __init__(self, raise_failure=False, *args, **kwargs):
 
     async def aio_extract(self, filename):
         """Run the extractor"""
+        filename_pathlib = Path(filename)
         # Resolve path in case of cwd change
-        filename = os.path.abspath(filename)
+        filename = str(filename_pathlib.resolve())
         for extractor in self.file_extractors:
             for extension in self.file_extractors[extractor]:
                 if filename.endswith(extension):
-                    extracted_path = os.path.join(
-                        self.tempdir, f"{os.path.basename(filename)}.extracted"
+                    extracted_path = str(
+                        Path(self.tempdir) / f"{filename_pathlib.name}.extracted"
                     )
-                    if os.path.exists(extracted_path):
+                    if Path(extracted_path).exists():
                         await aio_rmdir(extracted_path)
                     await aio_makedirs(extracted_path, 0o700)
                     async with ChangeDirContext(extracted_path):