address review comments

Signed-off-by: Tushar Goel <tushar.goel.dav@gmail.com>

Author: TG1999

requirements.txt

@@ -8,7 +8,7 @@ django-widget-tweaks>=1.4.8
 packageurl-python>=0.9.4
 binaryornot>=0.4.4
 GitPython>=3.1.17
-univers>=30.3.0
+univers>=30.3.1
 saneyaml>=0.5.2
 beautifulsoup4>=4.9.3
 python-dateutil>=2.8.1

vulnerabilities/importers/github.py

@@ -331,8 +331,6 @@ def process_response(resp: dict, package_type: str) -> Iterable[AdvisoryData]:
         )
         return
 
-    LOGGER.error(f"Found {len(vulnerabilities)} vulnerabilities for package_type: {package_type!r}")
-
     for vulnerability in vulnerabilities:
         affected_packages = []
         aliases = set()
@@ -458,6 +456,7 @@ def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
                 advisory_data.affected_packages
             )
         except UnMergeablePackageError:
+            LOGGER.error(f"Cannot merge with different purls {advisory_data.affected_packages!r}")
             return iter([])
 
         pkg_type = purl.type

vulnerabilities/importers/nginx.py

@@ -201,7 +201,9 @@ def get_inferences(self, advisory_data: AdvisoryData) -> Iterable[Inference]:
                 advisory_data.affected_packages
             )
         except UnMergeablePackageError:
+            logger.error(f"Cannot merge with different purls {advisory_data.affected_packages!r}")
             return iter([])
+
         all_versions = self.version_api.get("nginx/nginx").valid_versions
         affected_purls = []
         for affected_version_range in affected_version_ranges:

vulnerabilities/package_managers_2.py

@@ -3,28 +3,33 @@
 import traceback
 import xml.etree.ElementTree as ET
 from datetime import datetime
-from json import JSONDecodeError
 from typing import List
 from typing import MutableMapping
 from typing import Optional
 from typing import Set
+from urllib.parse import urlparse
 
 import requests
 from dateutil import parser as dateparser
 from django.utils.dateparse import parse_datetime
 
-from vulnerabilities.package_managers import Version as LegacyVersion
 from vulnerabilities.package_managers import VersionResponse
 
 LOGGER = logging.getLogger(__name__)
 
 
 @dataclasses.dataclass(frozen=True)
-class Version:
+class LegacyVersion:
     value: str
     release_date: Optional[datetime] = None
 
 
+@dataclasses.dataclass
+class VersionResponse:
+    valid_versions: Set[str] = dataclasses.field(default_factory=set)
+    newer_versions: Set[str] = dataclasses.field(default_factory=set)
+
+
 def get_response(url, type="json"):
     resp = requests.get(url=url)
     if not resp.status_code == 200:
@@ -81,12 +86,14 @@ def fetch(self, pkg):
             self.cache[pkg] = versions
             return
 
-        for version, download_items in response["releases"].items():
+        for version, download_items in response["releases"].items() or {}:
             if download_items:
                 latest_download_item = max(
                     download_items,
                     key=lambda download_item: dateparser.parse(
                         download_item["upload_time_iso_8601"]
+                        if "upload_time_iso_8601" in download_item
+                        else LOGGER.error(f"{download_item} has no upload_time_iso_8601")
                     ),
                 )
                 versions.add(
@@ -110,34 +117,11 @@ def fetch(self, pkg):
             self.cache[pkg] = versions
             return
         for release in response:
-            versions.add(
-                LegacyVersion(
-                    value=release["number"],
-                    release_date=dateparser.parse(release["created_at"]),
-                )
-            )
-
-        self.cache[pkg] = versions
-
-
-class NpmVersionAPI(VersionAPI):
-
-    package_type = "npm"
-
-    def fetch(self, pkg):
-        url = f"https://registry.npmjs.org/{pkg}"
-        versions = set()
-        response = get_response(url=url)
-        if not response:
-            self.cache[pkg] = versions
-            return
-        for version in response.get("versions", []):
-            release_date = response.get("time", {}).get(version)
-            if release_date:
-                release_date = dateparser.parse(release_date)
-                versions.add(LegacyVersion(value=version, release_date=release_date))
+            if release["number"] and release["published_at"]:
+                release_date = dateparser.parse(release["published_at"])
+                versions.add(LegacyVersion(value=release["number"], release_date=release_date))
             else:
-                versions.add(LegacyVersion(value=version, release_date=None))
+                LOGGER.error(f"Failed to parse release {release}")
 
         self.cache[pkg] = versions
 
@@ -213,19 +197,23 @@ def nuget_url(pkg_name: str) -> str:
     def extract_versions(resp: dict) -> Set[LegacyVersion]:
         all_versions = set()
         try:
-            for entry_group in resp["items"]:
-                for entry in entry_group["items"]:
-                    all_versions.add(
-                        LegacyVersion(
-                            value=entry["catalogEntry"]["version"],
-                            release_date=dateparser.parse(entry["catalogEntry"]["published"]),
+            for entry_group in resp["items"] or []:
+                for entry in entry_group["items"] or []:
+                    catalog_entry = entry["catalogEntry"] or {}
+                    version = catalog_entry.get("version")
+                    release_date = dateparser.parse(catalog_entry.get("published"))
+                    if version and release_date:
+                        all_versions.add(
+                            LegacyVersion(
+                                value=version,
+                                release_date=release_date,
+                            )
                         )
-                    )
         # FIXME: json response for YamlDotNet.Signed triggers this exception.
         # Some packages with many versions give a response of a list of endpoints.
         # In such cases rather, we should collect data from those endpoints.
         except KeyError:
-            pass
+            LOGGER.error(f"Failed to parse nuget response for {resp!r}")
 
         return all_versions
 
@@ -239,7 +227,7 @@ def __init__(self, cache: MutableMapping[str, Set[LegacyVersion]] = None):
         self.module_name_by_package_name = {}
 
     @staticmethod
-    def trim_url_path(url_path: str) -> Optional[str]:
+    def trim_go_url_path(url_path: str) -> Optional[str]:
         """
         Return a trimmed Go `url_path` removing trailing
         package references and keeping only the module
@@ -254,17 +242,21 @@ def trim_url_path(url_path: str) -> Optional[str]:
         This functions trims the trailing part(s) of a package URL
         and returns the remaining the module name.
         For example:
-        >>> module = "https://github.com/xx/a"
-        >>> assert GoproxyVersionAPI.trim_url_path("https://github.com/xx/a/b") == module
+        >>> module = "github.com/xx/a"
+        >>> assert GoproxyVersionAPI.trim_go_url_path("https://github.com/xx/a/b") == module
         """
         # some advisories contains this prefix in package name, e.g. https://github.com/advisories/GHSA-7h6j-2268-fhcm
         if url_path.startswith("https://pkg.go.dev/"):
-            url_path = url_path.removeprefix("https://pkg.go.dev/")
-        parts = url_path.split("/")
-        if len(parts) >= 2:
-            return "/".join(parts[:-1])
-        else:
+            url_path = url_path[len("https://pkg.go.dev/") :]
+        parsed_url_path = urlparse(url_path)
+        path = parsed_url_path.path
+        parts = path.split("/")
+        if len(parts) < 3:
+            LOGGER.error(f"Not a valid Go URL path {url_path} trim_go_url_path")
             return None
+        else:
+            joined_path = "/".join(parts[:3])
+            return f"{parsed_url_path.netloc}{joined_path}"
 
     @staticmethod
     def escape_path(path: str) -> str:
@@ -318,8 +310,8 @@ def fetch(self, pkg: str):
             url = f"https://proxy.golang.org/{escaped_pkg}/@v/list"
             resp_text = get_response(url=url, type="text")
             if not resp_text:
-                escaped_pkg = GoproxyVersionAPI.trim_url_path(escaped_pkg)
-                trimmed_pkg = GoproxyVersionAPI.trim_url_path(trimmed_pkg) or ""
+                escaped_pkg = GoproxyVersionAPI.trim_go_url_path(escaped_pkg)
+                trimmed_pkg = GoproxyVersionAPI.trim_go_url_path(trimmed_pkg) or ""
                 continue
             break
         if resp_text is None or escaped_pkg is None or trimmed_pkg is None:

vulnerabilities/tests/test_package_managers_2.py

@@ -0,0 +1,16 @@
+import pytest
+
+from vulnerabilities.package_managers_2 import GoproxyVersionAPI
+
+
+@pytest.mark.parametrize(
+    "url_path", ["https://pkg.go.dev/https://github.com/xx/a/b", "https://github.com/xx/a/b"]
+)
+def test_trim_go_url_path(url_path):
+    assert GoproxyVersionAPI.trim_go_url_path(url_path) == "github.com/xx/a"
+
+
+def test_trim_go_url_path_failure(caplog):
+    url_path = "https://github.com"
+    assert GoproxyVersionAPI.trim_go_url_path(url_path) == None
+    assert "Not a valid Go URL path" in caplog.text