Add PyPa live importer #1953

* Add PyPa live pipeline importer to fetch advisories affecting a single PURL

* Add tests for PyPa live importer

Signed-off-by: Michael Ehab Mikhail <michael.ehab@hotmail.com>

Author: michaelehab

vulnerabilities/importers/__init__.py

@@ -55,6 +55,7 @@
 from vulnerabilities.pipelines.v2_importers import oss_fuzz as oss_fuzz_v2
 from vulnerabilities.pipelines.v2_importers import postgresql_importer as postgresql_importer_v2
 from vulnerabilities.pipelines.v2_importers import pypa_importer as pypa_importer_v2
+from vulnerabilities.pipelines.v2_importers import pypa_live_importer as pypa_live_importer_v2
 from vulnerabilities.pipelines.v2_importers import pysec_importer as pysec_importer_v2
 from vulnerabilities.pipelines.v2_importers import vulnrichment_importer as vulnrichment_importer_v2
 from vulnerabilities.pipelines.v2_importers import xen_importer as xen_importer_v2
@@ -113,3 +114,9 @@
         oss_fuzz.OSSFuzzImporter,
     ]
 )
+
+LIVE_IMPORTERS_REGISTRY = create_registry(
+    [
+        pypa_live_importer_v2.PyPaLiveImporterPipeline,
+    ]
+)

vulnerabilities/pipelines/v2_importers/pypa_live_importer.py

@@ -0,0 +1,150 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+
+from typing import Iterable
+
+import requests
+import saneyaml
+from packageurl import PackageURL
+from univers.versions import PypiVersion
+
+from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+
+
+class PyPaLiveImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
+    """
+    Pypa Live Importer Pipeline
+
+    Collect advisories from PyPA GitHub repository for a single PURL.
+    """
+
+    pipeline_id = "pypa_live_importer_v2"
+    supported_types = ["pypi"]
+    spdx_license_expression = "CC-BY-4.0"
+    license_url = "https://github.com/pypa/advisory-database/blob/main/LICENSE"
+
+    @classmethod
+    def steps(cls):
+        return (
+            cls.get_purl_inputs,
+            cls.fetch_package_advisories,
+            cls.collect_and_store_advisories,
+        )
+
+    def get_purl_inputs(self):
+        purl = self.inputs["purl"]
+        if not purl:
+            raise ValueError("PURL is required for PyPaLiveImporterPipeline")
+
+        if isinstance(purl, str):
+            purl = PackageURL.from_string(purl)
+
+        if not isinstance(purl, PackageURL):
+            raise ValueError(f"Object of type {type(purl)} {purl!r} is not a PackageURL instance")
+
+        if purl.type not in self.supported_types:
+            raise ValueError(
+                f"PURL: {purl!s} is not among the supported package types {self.supported_types!r}"
+            )
+
+        if not purl.version:
+            raise ValueError(f"PURL: {purl!s} is expected to have a version")
+
+        self.purl = purl
+
+    def _is_version_affected(self, advisory_dict, version):
+        affected = advisory_dict.get("affected", [])
+        try:
+            v = PypiVersion(version)
+        except Exception:
+            return False
+        for entry in affected:
+            ranges = entry.get("ranges", [])
+            for r in ranges:
+                events = r.get("events", [])
+                introduced = None
+                fixed = None
+                for event in events:
+                    if "introduced" in event:
+                        introduced = event["introduced"]
+                    if "fixed" in event:
+                        fixed = event["fixed"]
+                try:
+                    if introduced:
+                        introduced_v = PypiVersion(introduced)
+                        if v < introduced_v:
+                            continue
+                    if fixed:
+                        fixed_v = PypiVersion(fixed)
+                        if v >= fixed_v:
+                            continue
+                    if introduced:
+                        introduced_v = PypiVersion(introduced)
+                        if (not fixed or v < PypiVersion(fixed)) and v >= introduced_v:
+                            return True
+                except Exception:
+                    continue
+        return False
+
+    def fetch_package_advisories(self):
+        if not self.purl.type in self.supported_types:
+            return
+
+        search_path = f"vulns/{self.purl.name}"
+
+        self.package_advisories = []
+
+        api_url = f"https://api.github.com/repos/pypa/advisory-database/contents/{search_path}"
+        response = requests.get(api_url)
+
+        if response.status_code == 404:
+            self.log(f"No advisories found for package {self.purl.name}")
+            return
+
+        if response.status_code != 200:
+            self.log(f"Failed to fetch advisories: {response.status_code} {response.text}")
+            return
+
+        for item in response.json():
+            if item["type"] == "file" and item["name"].endswith(".yaml"):
+                file_url = item["download_url"]
+                self.log("Fetching advisory file: " + item["name"])
+                file_response = requests.get(file_url)
+
+                if file_response.status_code == 200:
+                    advisory_text = file_response.text
+                    advisory_dict = saneyaml.load(advisory_text)
+
+                    if self.purl.version and not self._is_version_affected(
+                        advisory_dict, self.purl.version
+                    ):
+                        continue
+
+                    self.package_advisories.append(
+                        {"text": advisory_text, "dict": advisory_dict, "url": item["html_url"]}
+                    )
+
+    def advisories_count(self):
+        return len(self.package_advisories) if hasattr(self, "package_advisories") else 0
+
+    def collect_advisories(self) -> Iterable[AdvisoryData]:
+        from vulnerabilities.importers.osv import parse_advisory_data_v2
+
+        if not hasattr(self, "package_advisories"):
+            return
+
+        for advisory in self.package_advisories:
+            yield parse_advisory_data_v2(
+                raw_data=advisory["dict"],
+                supported_ecosystems=self.supported_types,
+                advisory_url=advisory["url"],
+                advisory_text=advisory["text"],
+            )

vulnerabilities/tests/pipelines/test_pypa_v2_live_importer_pipeline.py

@@ -0,0 +1,134 @@
+#
+# Copyright (c) nexB Inc. and others. All rights reserved.
+# VulnerableCode is a trademark of nexB Inc.
+# SPDX-License-Identifier: Apache-2.0
+# See http://www.apache.org/licenses/LICENSE-2.0 for the license text.
+# See https://github.com/aboutcode-org/vulnerablecode for support or download.
+# See https://aboutcode.org for more information about nexB OSS projects.
+#
+
+from unittest.mock import MagicMock
+from unittest.mock import patch
+
+import pytest
+import saneyaml
+from packageurl import PackageURL
+
+from vulnerabilities.importer import AdvisoryData
+
+
+@pytest.fixture
+def mock_github_api_response():
+    return {
+        "status_code": 200,
+        "json": [
+            {
+                "type": "file",
+                "name": "CVE-2022-1234.yaml",
+                "download_url": "https://raw.githubusercontent.com/pypa/advisory-database/main/vulns/package1/CVE-2022-1234.yaml",
+                "html_url": "https://github.com/pypa/advisory-database/blob/main/vulns/package1/CVE-2022-1234.yaml",
+            },
+            {
+                "type": "file",
+                "name": "CVE-2022-5678.yaml",
+                "download_url": "https://raw.githubusercontent.com/pypa/advisory-database/main/vulns/package1/CVE-2022-5678.yaml",
+                "html_url": "https://github.com/pypa/advisory-database/blob/main/vulns/package1/CVE-2022-5678.yaml",
+            },
+        ],
+    }
+
+
+@pytest.fixture
+def mock_advisory_files():
+    advisory1 = {
+        "id": "CVE-2022-1234",
+        "summary": "A vulnerability in package1",
+        "affected": [
+            {
+                "package": {"name": "package1", "ecosystem": "PyPI"},
+                "ranges": [
+                    {"type": "ECOSYSTEM", "events": [{"introduced": "1.0.0"}, {"fixed": "1.2.0"}]}
+                ],
+            }
+        ],
+    }
+
+    advisory2 = {
+        "id": "CVE-2022-5678",
+        "summary": "Another vulnerability in package1",
+        "affected": [
+            {
+                "package": {"name": "package1", "ecosystem": "PyPI"},
+                "ranges": [
+                    {"type": "ECOSYSTEM", "events": [{"introduced": "1.5.0"}, {"fixed": "1.7.0"}]}
+                ],
+            }
+        ],
+    }
+
+    return {
+        "https://raw.githubusercontent.com/pypa/advisory-database/main/vulns/package1/CVE-2022-1234.yaml": advisory1,
+        "https://raw.githubusercontent.com/pypa/advisory-database/main/vulns/package1/CVE-2022-5678.yaml": advisory2,
+    }
+
+
+def test_package_with_version_affected(mock_github_api_response, mock_advisory_files):
+    from vulnerabilities.pipelines.v2_importers.pypa_live_importer import PyPaLiveImporterPipeline
+
+    purl = PackageURL(type="pypi", name="package1", version="1.1.0")
+
+    with patch("requests.get") as mock_get:
+        mock_api_response = MagicMock()
+        mock_api_response.status_code = mock_github_api_response["status_code"]
+        mock_api_response.json.return_value = mock_github_api_response["json"]
+
+        def mock_get_side_effect(url, *args, **kwargs):
+            if "api.github.com" in url:
+                return mock_api_response
+
+            mock_file_response = MagicMock()
+            mock_file_response.status_code = 200
+            mock_file_response.text = saneyaml.dump(mock_advisory_files[url])
+            return mock_file_response
+
+        mock_get.side_effect = mock_get_side_effect
+
+        with patch("vulnerabilities.importers.osv.parse_advisory_data_v2") as mock_parse:
+
+            def side_effect(raw_data, supported_ecosystems, advisory_url, advisory_text):
+                return AdvisoryData(
+                    advisory_id=raw_data["id"],
+                    summary=raw_data["summary"],
+                    references_v2=[{"url": advisory_url}],
+                    affected_packages=[],
+                    weaknesses=[],
+                    url=advisory_url,
+                )
+
+            mock_parse.side_effect = side_effect
+
+            pipeline = PyPaLiveImporterPipeline(selected_groups=["package_first"], purl=purl)
+            pipeline.get_purl_inputs()
+            pipeline.fetch_package_advisories()
+            advisories = list(pipeline.collect_advisories())
+
+            assert len(advisories) == 1
+            assert advisories[0].advisory_id == "CVE-2022-1234"
+
+
+def test_nonexistent_package():
+    from vulnerabilities.pipelines.v2_importers.pypa_live_importer import PyPaLiveImporterPipeline
+
+    purl = PackageURL(type="pypi", name="nonexistent_package", version="1.0.0")
+
+    with patch("requests.get") as mock_get:
+        mock_response = MagicMock()
+        mock_response.status_code = 404
+        mock_get.return_value = mock_response
+
+        pipeline = PyPaLiveImporterPipeline(selected_groups=["package_first"], purl=purl)
+        pipeline.get_purl_inputs()
+        pipeline.fetch_package_advisories()
+        advisories = list(pipeline.collect_advisories())
+
+        assert len(advisories) == 0