extractor: Native Python support for rpm files (#995)

* setup: Reference requirements.txt for install_requires

setup.py and requirements.txt had slightly differing lists of
dependencies. By reading from requirements.txt in setup.py, we ensure
that they aways have the same list.

Signed-off-by: John Andersen <[email protected]>

* extractor: Native Python support for rpm files

Using rpmfile PyPi library we can extract rpm files if rpm2cpio and cpio
are not present.

Fixes: #657

Signed-off-by: John Andersen <[email protected]>

Author: John Andersen

MANIFEST.in

@@ -1,3 +1,4 @@
+include requirements.txt
 include *.md
 include test/binaries/*.c
 include test/csv/*.csv

cve_bin_tool/extractor.py

@@ -9,6 +9,8 @@
 import sys
 import tempfile
 
+from rpmfile.cli import main as rpmextract
+
 from .async_utils import (
     aio_unpack_archive,
     aio_run_command,
@@ -20,11 +22,16 @@
     ChangeDirContext,
     FileIO,
     run_coroutine,
+    async_wrap,
 )
 from .error_handler import ExtractionFailed, UnknownArchiveType, ErrorHandler, ErrorMode
 from .log import LOGGER
 
 
+# Run rpmfile in a thread
+rpmextract = async_wrap(rpmextract)
+
+
 class BaseExtractor:
     """Extracts tar, rpm, etc. files"""
 
@@ -66,10 +73,7 @@ async def extract_file_rpm(self, filename, extraction_path):
         """ Extract rpm packages """
         if sys.platform.startswith("linux"):
             if not await aio_inpath("rpm2cpio") or not await aio_inpath("cpio"):
-                with ErrorHandler(mode=self.error_mode, logger=self.logger):
-                    raise Exception(
-                        "'rpm2cpio' and 'cpio' are required to extract rpm files"
-                    )
+                await rpmextract("-xC", extraction_path, filename)
             else:
                 stdout, stderr = await aio_run_command(["rpm2cpio", filename])
                 if stderr or not stdout:

requirements.txt

@@ -1,12 +1,14 @@
-jsonschema
 rich
 plotly
 jinja2
 beautifulsoup4
 aiohttp[speedups]
 toml
 pyyaml
+jsonschema>=3.0.2
 pytest
 pytest-xdist
 pytest-cov
-pytest-asyncio
+pytest-asyncio
+rpmfile>=1.0.6
+zstandard; python_version >= "3.4"

setup.py

@@ -7,6 +7,9 @@
 with open("README.md", "r", encoding="utf-8") as f:
     readme = f.read()
 
+with open("requirements.txt", "r", encoding="utf-8") as f:
+    requirements = f.read().split("\n")
+
 with open(os.path.join("cve_bin_tool", "version.py"), "r") as f:
     for line in f:
         if line.startswith("VERSION"):
@@ -38,20 +41,7 @@
         "Programming Language :: Python :: Implementation :: CPython",
         "Programming Language :: Python :: Implementation :: PyPy",
     ],
-    install_requires=[
-        "rich",
-        "plotly",
-        "jinja2",
-        "beautifulsoup4",
-        "aiohttp[speedups]",
-        "toml",
-        "pyyaml",
-        "jsonschema>=3.0.2",
-        "pytest",
-        "pytest-xdist",
-        "pytest-cov",
-        "pytest-asyncio",
-    ],
+    install_requires=requirements,
     packages=find_packages(),
     package_data={
         "cve_bin_tool.output_engine": [

test/test_extractor.py

@@ -5,6 +5,7 @@
 import tarfile
 import tempfile
 import unittest
+import unittest.mock
 from io import BytesIO
 from zipfile import ZipFile, ZipInfo
 
@@ -102,8 +103,10 @@ async def test_extract_cleanup(self):
 class TestExtractFileRpm(TestExtractorBase):
     """ Tests for the rpm file extractor """
 
-    def setup_method(self):
-        download_file(CURL_7_20_0_URL, os.path.join(self.tempdir, "test.rpm"))
+    @classmethod
+    def setup_class(cls):
+        super(TestExtractFileRpm, cls).setup_class()
+        download_file(CURL_7_20_0_URL, os.path.join(cls.tempdir, "test.rpm"))
 
     @pytest.mark.asyncio
     async def test_extract_file_rpm(self):
@@ -116,6 +119,18 @@ async def test_extract_file_rpm(self):
         ):
             assert os.path.isfile(os.path.join(extracted_path, "usr", "bin", "curl"))
 
+    @pytest.mark.asyncio
+    async def test_extract_file_rpm_no_rpm2cipo(self):
+        """ Test rpm extraction using rpmfile """
+        with unittest.mock.patch(
+            "cve_bin_tool.async_utils.aio_inpath",
+            return_value=False,
+        ), unittest.mock.patch(
+            "cve_bin_tool.async_utils.aio_run_command",
+        ) as mock_aio_run_command:
+            await self.test_extract_file_rpm()
+            mock_aio_run_command.assert_not_called()
+
 
 class TestExtractFileDeb(TestExtractorBase):
     """ Tests for deb file extractor """