add fixed_filter with a test . fix aliases , purl

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

Author: ziadhany

vulnerabilities/importers/pysec.py

@@ -21,6 +21,7 @@
 #  Visit https://github.com/nexB/vulnerablecode/ for support and download.
 import json
 import logging
+from builtins import set
 from datetime import datetime
 from datetime import timezone
 from io import BytesIO
@@ -45,7 +46,7 @@
 
 
 class PyPIImporter(Importer):
-    spdx_license_expression = "Apache-2.0"
+    spdx_license_expression = "CC-BY-4.0"
 
     def advisory_data(self) -> Iterable[AdvisoryData]:
         """
@@ -55,32 +56,32 @@ def advisory_data(self) -> Iterable[AdvisoryData]:
         4. yield the json file to parse_advisory_data
         """
         url = "https://osv-vulnerabilities.storage.googleapis.com/PyPI/all.zip"
-        response = requests.get(url).content
         try:
+            response = requests.get(url).content
             with ZipFile(BytesIO(response)) as zip_file:
                 for file_name in zip_file.namelist():
                     with zip_file.open(file_name) as f:
                         vul_info = json.loads(f.read())
                         yield parse_advisory_data(vul_info)
         except requests.exceptions.RequestException:
-            logger.error(f"Failed to fetch osv-vulnerabilities PyPI: HTTP ")
+            logger.error("Failed to fetch osv-vulnerabilities PyPI")
 
 
 def parse_advisory_data(raw_data: dict) -> AdvisoryData:
-    if "summary" in raw_data:
-        summary = raw_data["summary"]
-    else:
-        summary = ""
+    summary = raw_data.get("summary") or ""
 
-    aliases = raw_data["aliases"] if "aliases" in raw_data else raw_data["id"]
+    vulnerability_id = raw_data.get("id")
+    vulnerability_aliases = raw_data.get("aliases") or []
+    vulnerability_aliases.append(vulnerability_id)
+    aliases = list(dict.fromkeys(vulnerability_aliases))
 
     if "published" in raw_data:
         date_published = datetime.strptime(
             raw_data["published"][0:19], "%Y-%m-%dT%H:%M:%S"
         ).replace(tzinfo=timezone.utc)
     else:
         date_published = None
-        logger.error("date_published not found " + raw_data["id"])
+        logger.warning("date_published not found " + raw_data["id"])
 
     severity = []
     if "severity" in raw_data:
@@ -92,7 +93,7 @@ def parse_advisory_data(raw_data: dict) -> AdvisoryData:
                         value=sever_list["score"],
                     )
                 )
-    elif "ecosystem_specific" in raw_data and "severity" in raw_data["ecosystem_specific"]:
+    if "ecosystem_specific" in raw_data and "severity" in raw_data["ecosystem_specific"]:
         severity.append(
             VulnerabilitySeverity(
                 system=SCORING_SYSTEMS["generic_textual"],
@@ -113,59 +114,59 @@ def parse_advisory_data(raw_data: dict) -> AdvisoryData:
     if "affected" in raw_data:
         for affected_pkg in raw_data["affected"]:
             package = affected_pkg["package"]
+
+            purl = ""
             if "ecosystem" in package and "name" in package:
                 purl = PackageURL(type=package["ecosystem"], name=package["name"])
+            if "purl" in package:
+                try:
+                    purl = PackageURL.from_string(package["purl"])
+                except ValueError:
+                    logger.error(
+                        f"PackageURL ValueError - {raw_data['id'] !r} - purl: {package['purl'] !r}"
+                    )
             else:
-                purl = ""
                 logger.error(f"purl affected_pkg not found - {raw_data['id'] !r}")
 
             affected_version_range = None
-            if "versions" in affected_pkg:
+            if "versions" in affected_pkg and len(affected_pkg["versions"]) < 200:
                 affected_version_range = (
-                    # HINT:  Values larger than 1/3 of a buffer page cannot be indexed.
-                    PypiVersionRange(affected_pkg["versions"][slice(200)])
+                    # HINT: Values larger than 1/3 of a buffer page cannot be indexed.
+                    PypiVersionRange(affected_pkg["versions"])
                 )
 
-            if "ranges" in affected_pkg:
-                for fix_range in affected_pkg["ranges"]:
-                    fixed_version = set()
-                    if "type" in fix_range:
-                        if fix_range["type"] == "ECOSYSTEM":
-                            filter_fixed = list(
-                                filter(lambda x: x.keys() == {"fixed"}, fix_range["events"])
-                            )
-                            list_fixed = [i["fixed"] for i in filter_fixed]
-                            for i in list_fixed:
-                                try:
-                                    fixed_version.add(PypiVersion(i))
-                                except InvalidVersion:
-                                    logger.error(
-                                        f"fixed_version InvalidVersion - PypiVersion - {raw_data['id'] !r}"
-                                    )
-
-                        if fix_range["type"] == "SEMVER":
-                            filter_fixed = list(
-                                filter(lambda x: x.keys() == {"fixed"}, fix_range["events"])
-                            )
-                            list_fixed = [i["fixed"] for i in filter_fixed]
-
-                            for i in list_fixed:
-                                try:
-                                    fixed_version.add(SemverVersion(i))
-                                except InvalidVersion:
-                                    logger.error(
-                                        f"fixed_version InvalidVersion - SemverVersion - {raw_data['id'] !r}"
-                                    )
+            for fixed_range in affected_pkg.get("ranges", []):
+                fixed_version = set()
+                if "type" in fixed_range:
+                    list_fixed = fixed_filter(fixed_range)
+                    for i in list_fixed:
+                        if fixed_range["type"] == "ECOSYSTEM":
+                            try:
+                                fixed_version.add(PypiVersion(i))
+                            except InvalidVersion:
+                                logger.error(
+                                    f"Invalid Version - PypiVersion - {raw_data['id'] !r} - {i !r}"
+                                )
+                        if fixed_range["type"] == "SEMVER":
+                            try:
+                                fixed_version.add(SemverVersion(i))
+                            except InvalidVersion:
+                                logger.error(
+                                    f"Invalid Version - SemverVersion - {raw_data['id'] !r} - {i !r}"
+                                )
 
                         # TODO add GitHubVersion univers fix_version
-                    for version in fixed_version:
-                        affected_package.append(
-                            AffectedPackage(
-                                package=purl,
-                                affected_version_range=affected_version_range,
-                                fixed_version=version,
-                            )
+                else:
+                    logger.error(f"Invalid type - {raw_data['id'] !r}")
+
+                for version in fixed_version:
+                    affected_package.append(
+                        AffectedPackage(
+                            package=purl,
+                            affected_version_range=affected_version_range,
+                            fixed_version=version,
                         )
+                    )
     else:
         logger.error(f"affected_package not found - {raw_data['id'] !r}")
 
@@ -176,3 +177,9 @@ def parse_advisory_data(raw_data: dict) -> AdvisoryData:
         references=references,
         date_published=date_published,
     )
+
+
+def fixed_filter(fixed_range) -> []:
+    filter_fixed = list(filter(lambda x: x.keys() == {"fixed"}, fixed_range["events"]))
+    list_fixed = [i["fixed"] for i in filter_fixed]
+    return list_fixed