refactor: Refactor the GithubOSVLiveImporterPipeline pipeline and add a test using real data

Signed-off-by: ziad hany <ziadhany2016@gmail.com>

Author: ziadhany

vulnerabilities/importers/__init__.py

@@ -119,7 +119,6 @@
         retiredotnet_importer_v2.RetireDotnetImporterPipeline,
         ubuntu_osv_importer_v2.UbuntuOSVImporterPipeline,
         alpine_linux_importer_v2.AlpineLinuxImporterPipeline,
-        nvd_importer.NVDImporterPipeline,
         github_importer.GitHubAPIImporterPipeline,
         gitlab_importer.GitLabImporterPipeline,
         github_osv.GithubOSVImporter,

vulnerabilities/pipelines/v2_importers/github_osv_live_importer.py

@@ -1,13 +1,24 @@
 import json
-from typing import Iterable
-from typing import Optional
 
+import dateparser
 import requests
 from packageurl import PackageURL
 from univers.version_range import RANGE_CLASS_BY_SCHEMES
 
-from vulnerabilities.importer import AdvisoryData
 from vulnerabilities.pipelines import VulnerableCodeBaseImporterPipelineV2
+from vulnerabilities.pipes.osv_v2 import parse_advisory_data_v3
+from vulnerabilities.utils import fetch_response
+
+ECOSYSTEM_BY_PURL_TYPE = {
+    "pypi": "PyPI",
+    "npm": "npm",
+    "maven": "Maven",
+    "composer": "Packagist",
+    "hex": "Hex",
+    "gem": "RubyGems",
+    "nuget": "NuGet",
+    "cargo": "crates.io",
+}
 
 
 class GithubOSVLiveImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
@@ -26,11 +37,12 @@ class GithubOSVLiveImporterPipeline(VulnerableCodeBaseImporterPipelineV2):
     def steps(cls):
         return (
             cls.get_purl_inputs,
+            cls.get_osv_advisories_urls,
             cls.collect_and_store_advisories,
         )
 
     def get_purl_inputs(self):
-        purl = self.inputs["purl"]
+        purl = self.inputs.get("purl")
         if not purl:
             raise ValueError("PURL is required for GithubOSVLiveImporterPipeline")
 
@@ -51,106 +63,94 @@ def get_purl_inputs(self):
         self.purl = purl
 
     def advisories_count(self):
-        self.advisories = fetch_github_osv_advisories_for_purl(self.purl)
-        return len(self.advisories)
-
-    def collect_advisories(self) -> Iterable[AdvisoryData]:
-        from vulnerabilities.importers.osv import parse_advisory_data_v2
-
-        supported_ecosystems = [
-            "pypi",
-            "npm",
-            "maven",
-            # "golang",
-            "composer",
-            "hex",
-            "gem",
-            "nuget",
-            "cargo",
-        ]
-
-        input_version = self.purl.version
-        vrc = RANGE_CLASS_BY_SCHEMES[self.purl.type]
-        version_obj = vrc.version_class(input_version)
-
-        for adv in self.advisories:
-            adv_id = adv.get("id")
-            advisory_url = build_github_repo_advisory_url(adv, adv_id)
-
-            advisory = parse_advisory_data_v2(
-                raw_data=adv,
-                supported_ecosystems=supported_ecosystems,
+        return len(self.advisory_urls)
+
+    def collect_advisories(self):
+        """
+        Fetch and parse advisory data from GitHub Advisory Database URLs, Filters the packages to
+        ensure they match the exact type, name, and namespace of the target PURL, and ensure the target
+        version falls within the affected or fixed version ranges and yield these related advisories
+        """
+        version_range = RANGE_CLASS_BY_SCHEMES.get(self.purl.type)
+        version_obj = version_range.version_class(self.purl.version)
+        for advisory_url in self.advisory_urls:
+            response = fetch_response(advisory_url)
+            raw_data = json.loads(response.content)
+
+            advisory = parse_advisory_data_v3(
+                raw_data=raw_data,
+                supported_ecosystems=self.supported_types,
                 advisory_url=advisory_url,
-                advisory_text=json.dumps(adv, ensure_ascii=False),
+                advisory_text=json.dumps(raw_data, ensure_ascii=False),
             )
 
-            advisory.affected_packages = [
-                ap
-                for ap in advisory.affected_packages
-                if ap.package
-                and ap.package.type == self.purl.type
-                and ap.package.name == self.purl.name
-                and (ap.package.namespace or "") == (self.purl.namespace or "")
+            filtered_affected_packages = [
+                affected_package
+                for affected_package in advisory.affected_packages
+                if affected_package.package
+                and affected_package.package.type == self.purl.type
+                and affected_package.package.name == self.purl.name
+                and (affected_package.package.namespace or "") == (self.purl.namespace or "")
             ]
 
-            if not advisory.affected_packages:
+            if not filtered_affected_packages:
                 continue
 
-            if any(
-                ap.affected_version_range and version_obj in ap.affected_version_range
-                for ap in advisory.affected_packages
-            ):
-                yield advisory
-
-
-ECOSYSTEM_BY_PURL_TYPE = {
-    "pypi": "PyPI",
-    "npm": "npm",
-    "maven": "Maven",
-    "composer": "Packagist",
-    "hex": "Hex",
-    "gem": "RubyGems",
-    "nuget": "NuGet",
-    "cargo": "crates.io",
-}
-
-# Map purl.type to directory names used in the advisory-database repository
-REPO_DIR_BY_PURL_TYPE = {
-    "pypi": "pypi",
-    "npm": "npm",
-    "maven": "maven",
-    "composer": "composer",
-    "hex": "hex",
-    "gem": "rubygems",
-    "nuget": "nuget",
-    "cargo": "crates.io",
-}
+            for affected_package in filtered_affected_packages:
+                if (
+                    affected_package.affected_version_range
+                    and version_obj in affected_package.affected_version_range
+                ) or (
+                    affected_package.fixed_version_range
+                    and version_obj in affected_package.fixed_version_range
+                ):
+                    yield advisory
+
+    def get_osv_advisories_urls(self):
+        """
+        Fetch a list of OSV advisory dicts from the OSV API for a given PURL,
+        filtered to only GitHub advisories (GHSA-*) and return the Advisories URLS.
+        """
+        ecosystem = ECOSYSTEM_BY_PURL_TYPE.get(self.purl.type)
+        if not ecosystem:
+            return []
 
+        # Query by package to get all advisories for that package; we filter GHSA below.
+        body = {"package": {"ecosystem": ecosystem, "name": _osv_package_name(self.purl)}}
+        resp = requests.post("https://api.osv.dev/v1/query", json=body, timeout=30)
+        if resp.status_code != 200:
+            return []
 
-def build_github_repo_advisory_url(adv: dict, adv_id: Optional[str]) -> str:
+        data = resp.json() or {}
+        advisories = data.get("vulns") or []
+        self.advisory_urls = set()
+        for advisory in advisories:
+            adv_id = advisory.get("id") or ""
+            aliases = advisory.get("aliases") or []
+            advisory_ids = [adv_id] + aliases
+            for ghsa_id in advisory_ids:
+                if not ghsa_id.startswith("GHSA-"):
+                    continue
+
+                published_date = advisory.get("published")
+                advisory_url = build_github_repo_advisory_url(
+                    published_date, ghsa_id, logger=self.log
+                )
+                self.advisory_urls.add(advisory_url)
+
+
+def build_github_repo_advisory_url(published_date, advisory_id, logger):
     """
     Return the advisory JSON URL in the GitHub advisory-database repo, using the GHSA path:
     advisories/github-reviewed/YYYY/MM/GHSA-ID/GHSA-ID.json
     """
-    base = "https://github.com/github/advisory-database/blob/main/advisories/github-reviewed"
-    if not adv_id:
-        return f"{base}/"
-
-    date_str = adv.get("published") or adv.get("modified")
+    if not published_date:
+        logger(f"Cannot build URL for {advisory_id}: Missing both published and modified dates")
 
-    if date_str:
-        from datetime import datetime
-
-        try:
-            dt = datetime.fromisoformat(date_str.replace("Z", "+00:00"))
-            year = dt.strftime("%Y")
-            month = dt.strftime("%m")
-            return f"{base}/{year}/{month}/{adv_id}/{adv_id}.json"
-        except Exception:
-            pass
-
-    # Fallback to the base directory if no parseable date is present
-    return f"{base}/"
+    parsed_date = dateparser.parse(date_string=published_date)
+    year = parsed_date.strftime("%Y")
+    month = parsed_date.strftime("%m")
+    return f"https://raw.githubusercontent.com/github/advisory-database/refs/heads/main/advisories/github-reviewed/{year}/{month}/{advisory_id}/{advisory_id}.json"
 
 
 def _osv_package_name(purl: PackageURL) -> str:
@@ -160,27 +160,3 @@ def _osv_package_name(purl: PackageURL) -> str:
     if purl.namespace:
         return f"{purl.namespace}/{purl.name}"
     return purl.name
-
-
-def fetch_github_osv_advisories_for_purl(purl: PackageURL):
-    """
-    Return a list of OSV advisory dicts from the OSV API for a given PURL,
-    filtered to only GitHub advisories (GHSA-*).
-    """
-    ecosystem = ECOSYSTEM_BY_PURL_TYPE.get(purl.type)
-    if not ecosystem:
-        return []
-
-    pkg = {"ecosystem": ecosystem, "name": _osv_package_name(purl)}
-    # Query by package to get all advisories for that package; we filter GHSA below.
-    body = {"package": pkg}
-    try:
-        resp = requests.post("https://api.osv.dev/v1/query", json=body, timeout=30)
-        if resp.status_code != 200:
-            return []
-        data = resp.json() or {}
-        vulns = data.get("vulns") or []
-        # Keep only GHSA advisories which correspond to GitHub Advisory Database
-        return [v for v in vulns if isinstance(v.get("id"), str) and v["id"].startswith("GHSA-")]
-    except Exception:
-        return []

vulnerabilities/tests/pipelines/v2_importers/test_github_osv_live_importer_v2.py

@@ -1,59 +1,63 @@
 import json
+from pathlib import Path
 from unittest import mock
 
+import pytest
 from packageurl import PackageURL
 
-from vulnerabilities.importer import AdvisoryData
+from vulnerabilities.models import AdvisoryV2
 from vulnerabilities.pipelines.v2_importers.github_osv_live_importer import (
     GithubOSVLiveImporterPipeline,
 )
+from vulnerabilities.pipelines.v2_importers.github_osv_live_importer import (
+    build_github_repo_advisory_url,
+)
+from vulnerabilities.tests import util_tests
 
-SAMPLE_OSV = {
-    "id": "GHSA-xxxx-yyyy-zzzz",
-    "summary": "Sample summary",
-    "details": "Sample details",
-    "aliases": ["CVE-2021-99999"],
-    "affected": [
-        {
-            "package": {"name": "sample", "ecosystem": "PyPI"},
-            "ranges": [
-                {"type": "ECOSYSTEM", "events": [{"introduced": "1.0.0"}, {"fixed": "1.2.0"}]}
-            ],
-            "versions": ["1.0.0", "1.1.0"],
-        }
-    ],
-    "database_specific": {"cwe_ids": ["CWE-79"]},
-}
+TEST_DATA = Path(__file__).parent.parent.parent / "test_data" / "live_github_osv"
 
 
-@mock.patch(
-    "vulnerabilities.pipelines.v2_importers.github_osv_live_importer.fetch_github_osv_advisories_for_purl"
-)
-def test_github_osv_live_importer_found_with_version(mock_fetch):
-    mock_fetch.return_value = [json.loads(json.dumps(SAMPLE_OSV))]
-    purl = PackageURL(type="pypi", name="sample", version="1.1.0")
+@pytest.mark.django_db
+@mock.patch("vulnerabilities.pipelines.v2_importers.github_osv_live_importer.fetch_response")
+@mock.patch("vulnerabilities.pipelines.v2_importers.github_osv_live_importer.requests.post")
+def test_github_osv_live_importer(mocker_osv, mock_github_osv):
+    purl = PackageURL(type="pypi", name="django", version="1.4.2")
+
+    mocker_osv.return_value.status_code = 200
+    osv_api_path = TEST_DATA / "fetch_osv_api.json"
+    with open(osv_api_path, encoding="utf-8") as f:
+        mocker_osv.return_value.json.return_value = json.load(f)
+
+    github_osv_path = TEST_DATA / "fetch_github_osv.json"
+    with open(github_osv_path, encoding="utf-8") as f:
+        raw_advisory_list = json.load(f)
+
+    mock_github_osv.side_effect = lambda url: mock.Mock(
+        content=json.dumps(next(adv for adv in raw_advisory_list if adv.get("id") in url))
+    )
+
     pipeline = GithubOSVLiveImporterPipeline(purl=purl)
-    pipeline.get_purl_inputs()
-    pipeline.advisories_count()
-    advisories = list(pipeline.collect_advisories())
-    assert len(advisories) == 1
-    adv = advisories[0]
-    assert isinstance(adv, AdvisoryData)
-    assert adv.advisory_id == "GHSA-xxxx-yyyy-zzzz"
-    assert "CVE-2021-99999" in adv.aliases
-    assert adv.summary.startswith("Sample")
-    assert adv.affected_packages
-    assert adv.affected_packages[0].package.type == "pypi"
-
-
-@mock.patch(
-    "vulnerabilities.pipelines.v2_importers.github_osv_live_importer.fetch_github_osv_advisories_for_purl"
+    pipeline.execute()
+
+    expected_file = TEST_DATA / "expected-advisories.json"
+    result = [adv.to_advisory_data().to_dict() for adv in AdvisoryV2.objects.all()]
+    util_tests.check_results_against_json(result, expected_file)
+
+
+@pytest.mark.parametrize(
+    "published_date, advisory_id, expected_url",
+    [
+        (
+            "2022-05-17T05:10:31Z",
+            "GHSA-2655-q453-22f9",
+            "https://raw.githubusercontent.com/github/advisory-database/refs/heads/main/advisories/github-reviewed/2022/05/GHSA-2655-q453-22f9/GHSA-2655-q453-22f9.json",
+        ),
+        (
+            "2017-10-24T18:33:37Z",
+            "GHSA-4936-rj25-6wm6",
+            "https://raw.githubusercontent.com/github/advisory-database/refs/heads/main/advisories/github-reviewed/2017/10/GHSA-4936-rj25-6wm6/GHSA-4936-rj25-6wm6.json",
+        ),
+    ],
 )
-def test_github_osv_live_importer_none_found_with_version(mock_fetch):
-    mock_fetch.return_value = [json.loads(json.dumps(SAMPLE_OSV))]
-    purl = PackageURL(type="pypi", name="sample", version="1.2.0")
-    pipeline = GithubOSVLiveImporterPipeline(purl=purl)
-    pipeline.get_purl_inputs()
-    pipeline.advisories_count()
-    advisories = list(pipeline.collect_advisories())
-    assert advisories == []
+def test_build_github_repo_advisory_url(published_date, advisory_id, expected_url):
+    assert build_github_repo_advisory_url(published_date, advisory_id, logger=print) == expected_url