Format and streamline OSV and PyPA code

* Use named arguments when calling functions.
* Use variables for tests expected and results
* Split tests functions in smaller functions

Signed-off-by: Philippe Ombredanne <[email protected]>

Author: pombredanne

vulnerabilities/importers/osv.py

@@ -32,41 +32,39 @@
 
 
 def parse_advisory_data(raw_data: dict, supported_ecosystem) -> Optional[AdvisoryData]:
+    """
+    Return an AdvisoryData build from a ``raw_data`` mapping of OSV advisory and
+    a ``supported_ecosystem`` string.
+    """
     raw_id = raw_data.get("id") or ""
     summary = raw_data.get("summary") or ""
     details = raw_data.get("details") or ""
     summary = build_description(summary=summary, description=details)
     aliases = raw_data.get("aliases") or []
     if raw_id:
         aliases.append(raw_id)
-        aliases = dedupe(aliases)
-    date_published = get_published_date(raw_data)
-    severity = list(get_severities(raw_data))
-    references = get_references(raw_data, severity)
+        aliases = dedupe(original=aliases)
+
+    date_published = get_published_date(raw_data=raw_data)
+    severities = list(get_severities(raw_data=raw_data))
+    references = get_references(raw_data=raw_data, severities=severities)
 
     affected_packages = []
-    if "affected" not in raw_data:
-        return AdvisoryData(
-            aliases=aliases,
-            summary=summary,
-            references=references,
-            affected_packages=[],
-            date_published=date_published,
-        )
 
     for affected_pkg in raw_data.get("affected") or []:
-        purl = get_affected_purl(affected_pkg, raw_id)
+        purl = get_affected_purl(affected_pkg=affected_pkg, raw_id=raw_id)
         if purl.type != supported_ecosystem:
-            logger.error(
-                f"un supported ecosystem package found in the advisories: {purl} - from: {raw_id !r}"
-            )
+            logger.error(f"Unsupported package type: {purl!r} in OSV: {raw_id!r}")
             continue
 
         affected_version_range = get_affected_version_range(
-            affected_pkg, raw_id, supported_ecosystem
+            affected_pkg=affected_pkg,
+            raw_id=raw_id,
+            supported_ecosystem=supported_ecosystem,
         )
-        for fixed_range in affected_pkg.get("ranges", []):
-            fixed_version = get_fixed_version(fixed_range, raw_id)
+
+        for fixed_range in affected_pkg.get("ranges") or []:
+            fixed_version = get_fixed_versions(fixed_range=fixed_range, raw_id=raw_id)
 
             for version in fixed_version:
                 affected_packages.append(
@@ -80,18 +78,24 @@ def parse_advisory_data(raw_data: dict, supported_ecosystem) -> Optional[Advisor
     return AdvisoryData(
         aliases=aliases,
         summary=summary,
-        affected_packages=affected_packages,
         references=references,
+        affected_packages=affected_packages,
         date_published=date_published,
     )
 
 
-def fixed_filter(fixed_range) -> Iterable[str]:
+def extract_fixed_versions(fixed_range) -> Iterable[str]:
     """
-    Return a list of fixed version strings given a ``fixed_range`` mapping of OSV data.
-    >>> list(fixed_filter({"type": "SEMVER", "events": [{"introduced": "0"}, {"fixed": "1.6.0"}]}))
+    Return a list of fixed version strings given a ``fixed_range`` mapping of
+    OSV data.
+
+    >>> list(extract_fixed_versions(
+    ... {"type": "SEMVER", "events": [{"introduced": "0"},{"fixed": "1.6.0"}]}))
     ['1.6.0']
-    >>> list(fixed_filter({"type": "ECOSYSTEM","events":[{"introduced": "0"},{"fixed": "1.0.0"},{"fixed": "9.0.0"}]}))
+
+    >>> list(extract_fixed_versions(
+    ... {"type": "ECOSYSTEM","events":[{"introduced": "0"},
+    ... {"fixed": "1.0.0"},{"fixed": "9.0.0"}]}))
     ['1.0.0', '9.0.0']
     """
     for event in fixed_range.get("events") or []:
@@ -102,97 +106,141 @@ def fixed_filter(fixed_range) -> Iterable[str]:
 
 def get_published_date(raw_data):
     published = raw_data.get("published")
-    return published and dateparser.parse(published)
+    return published and dateparser.parse(date_string=published)
 
 
 def get_severities(raw_data) -> Iterable[VulnerabilitySeverity]:
-    for sever_list in raw_data.get("severity") or []:
-        if sever_list.get("type") == "CVSS_V3":
+    """
+    Yield VulnerabilitySeverity extracted from a mapping of OSV ``raw_data``
+    """
+    for severity in raw_data.get("severity") or []:
+        if severity.get("type") == "CVSS_V3":
             yield VulnerabilitySeverity(
-                system=SCORING_SYSTEMS["cvssv3.1_vector"], value=sever_list["score"]
+                system=SCORING_SYSTEMS["cvssv3.1_vector"],
+                value=severity["score"],
             )
         else:
-            logger.error(f"NotImplementedError severity type- {raw_data['id']!r}")
+            logger.error(f"Unsupported severity type: {severity!r} for OSV id: {raw_data['id']!r}")
 
-    ecosys = raw_data.get("ecosystem_specific") or {}
-    sever = ecosys.get("severity")
-    if sever:
+    ecosystem_specific = raw_data.get("ecosystem_specific") or {}
+    severity = ecosystem_specific.get("severity")
+    if severity:
         yield VulnerabilitySeverity(
             system=SCORING_SYSTEMS["generic_textual"],
-            value=sever,
+            value=severity,
         )
 
     database_specific = raw_data.get("database_specific") or {}
-    sever = database_specific.get("severity")
-    if sever:
+    severity = database_specific.get("severity")
+    if severity:
         yield VulnerabilitySeverity(
             system=SCORING_SYSTEMS["generic_textual"],
-            value=sever,
+            value=severity,
         )
 
 
 def get_references(raw_data, severities) -> List[Reference]:
-    references = raw_data.get("references") or []
-    return [Reference(url=ref["url"], severities=severities) for ref in references if ref]
+    """
+    Return a list Reference extracted from a mapping of OSV ``raw_data`` given a
+    ``severities`` list of VulnerabilitySeverity.
+    """
+    references = []
+    for ref in raw_data.get("references") or []:
+        if not ref:
+            continue
+
+        url = ref["url"]
+        if not url:
+            logger.error(f"Reference without URL : {ref!r} for OSV id: {raw_data['id']!r}")
+            continue
+        references.append(Reference(url=ref["url"], severities=severities))
+    return references
 
 
 def get_affected_purl(affected_pkg, raw_id):
+    """
+    Return an affected PackageURL or None given a mapping of ``affected_pkg``
+    data and a ``raw_id``.
+    """
     package = affected_pkg.get("package") or {}
     purl = package.get("purl")
     if purl:
         try:
             return PackageURL.from_string(purl)
         except ValueError:
-            logger.error(f"PackageURL ValueError - {raw_id !r} - purl: {purl !r}")
+            logger.error(
+                f"Invalid PackageURL: {purl!r} for OSV "
+                f"affected_pkg {affected_pkg} and id: {raw_id}"
+            )
 
     ecosys = package.get("ecosystem")
     name = package.get("name")
     if ecosys and name:
         return PackageURL(type=ecosys, name=name)
-    else:
-        logger.error(f"purl affected_pkg not found - {raw_id !r}")
+
+    logger.error(
+        f"No PackageURL possible: {purl!r} for affected_pkg {affected_pkg} for OSV id: {raw_id}"
+    )
 
 
 def get_affected_version_range(affected_pkg, raw_id, supported_ecosystem):
+    """
+    Return a univers VersionRange for the ``affected_pkg`` package data mapping
+    or None. Use a ``raw_id`` OSV id and ``supported_ecosystem``.
+    """
     affected_versions = affected_pkg.get("versions")
     if affected_versions:
         try:
             return RANGE_CLASS_BY_SCHEMES[supported_ecosystem].from_versions(affected_versions)
         except Exception as e:
             logger.error(
-                f"InvalidVersionRange affected_pkg_version_range Error - {raw_id !r} {e!r}"
+                f"Invalid VersionRange  for affected_pkg: {affected_pkg} "
+                f"for OSV id: {raw_id!r}: error:{e!r}"
             )
-    # else:
-    #     logger.error(f"affected_pkg_version_range not found - {raw_id !r} ")
 
 
-def get_fixed_version(fixed_range, raw_id) -> List[Version]:
+def get_fixed_versions(fixed_range, raw_id) -> List[Version]:
     """
-    Return a list of fixed versions, using fixed_filter we get the list of fixed version strings,
-    then we pass every element to their univers.versions , then we dedupe the result
-    >>> get_fixed_version({}, "GHSA-j3f7-7rmc-6wqj")
+    Return a list of unique fixed univers Versions given a ``fixed_range``
+    univers VersionRange and a ``raw_id``.
+
+    For example::
+
+    >>> get_fixed_versions(fixed_range={}, raw_id="GHSA-j3f7-7rmc-6wqj")
     []
-    >>> get_fixed_version({"type": "ECOSYSTEM", "events": [{"fixed": "1.7.0"}]}, "GHSA-j3f7-7rmc-6wqj")
+    >>> get_fixed_versions(
+    ...   fixed_range={"type": "ECOSYSTEM", "events": [{"fixed": "1.7.0"}]},
+    ...   raw_id="GHSA-j3f7-7rmc-6wqj"
+    ... )
     [PypiVersion(string='1.7.0')]
     """
-    fixed_version = []
+    fixed_versions = []
     if "type" not in fixed_range:
-        logger.error(f"Invalid type - {raw_id!r}")
-    else:
-        list_fixed = fixed_filter(fixed_range)
-        fixed_range_type = fixed_range["type"]
-        for i in list_fixed:
-            if fixed_range_type == "ECOSYSTEM":
-                try:
-                    fixed_version.append(PypiVersion(i))
-                except InvalidVersion:
-                    logger.error(f"Invalid Version - PypiVersion - {raw_id !r} - {i !r}")
-            if fixed_range_type == "SEMVER":
-                try:
-                    fixed_version.append(SemverVersion(i))
-                except InvalidVersion:
-                    logger.error(f"Invalid Version - SemverVersion - {raw_id !r} - {i !r}")
-            # if fixed_range_type == "GIT":
-            # TODO add GitHubVersion univers fix_version
-            #     logger.error(f"NotImplementedError GIT Version - {raw_id !r} - {i !r}")
-    return dedupe(fixed_version)
+        logger.error(f"Invalid fixed_range type for: {fixed_range} for OSV id: {raw_id!r}")
+        return []
+
+    fixed_range_type = fixed_range["type"]
+
+    for version in extract_fixed_versions(fixed_range):
+
+        # FIXME: ECOSYSTEM does not imply PyPI!!!!
+        if fixed_range_type == "ECOSYSTEM":
+            try:
+                fixed_versions.append(PypiVersion(version))
+            except InvalidVersion:
+                logger.error(f"Invalid PypiVersion: {version!r} for OSV id: {raw_id!r}")
+
+        elif fixed_range_type == "SEMVER":
+            try:
+                fixed_versions.append(SemverVersion(version))
+            except InvalidVersion:
+                logger.error(f"Invalid SemverVersion: {version!r} for OSV id: {raw_id!r}")
+
+        else:
+            logger.error(f"Unsupported fixed version type: {version!r} for OSV id: {raw_id!r}")
+
+        # if fixed_range_type == "GIT":
+        # TODO add GitHubVersion univers fix_version
+        #     logger.error(f"NotImplementedError GIT Version - {raw_id !r} - {i !r}")
+
+    return dedupe(fixed_versions)

vulnerabilities/importers/pypa.py

@@ -26,8 +26,8 @@ class PyPaImporter(Importer):
     url = "git+https://github.com/pypa/advisory-database"
 
     def advisory_data(self) -> Iterable[AdvisoryData]:
-        for file in fork_and_get_files(self.url):
-            yield parse_advisory_data(file, supported_ecosystem="pypi")
+        for raw_data in fork_and_get_files(self.url):
+            yield parse_advisory_data(raw_data=raw_data, supported_ecosystem="pypi")
 
 
 class ForkError(Exception):
@@ -36,20 +36,20 @@ class ForkError(Exception):
 
 def fork_and_get_files(url) -> dict:
     """
-    Fetch the github repository and go to vulns directory ,
-    then open directories one by one and return a file .
+    Yield advisorie data mappings from the PyPA GitHub repository at ``url``.
     """
     try:
         fork_directory = fetch_via_git(url=url)
     except Exception as e:
-        logger.error(f"Can't clone url {url}")
+        logger.error(f"Failed to clone url {url}: {e}")
         raise ForkError(url) from e
 
     advisory_dirs = os.path.join(fork_directory.dest_dir, "vulns")
     for root, _, files in os.walk(advisory_dirs):
         for file in files:
+            path = os.path.join(root, file)
             if not file.endswith(".yaml"):
-                logger.warning(f"unsupported file {file}")
-            else:
-                with open(os.path.join(root, file), "r") as f:
-                    yield saneyaml.load(f.read())
+                logger.warning(f"Unsupported non-YAML PyPA advisory file: {path}")
+                continue
+            with open(path) as f:
+                yield saneyaml.load(f.read())

vulnerabilities/importers/pysec.py

@@ -27,18 +27,15 @@ class PyPIImporter(Importer):
 
     def advisory_data(self) -> Iterable[AdvisoryData]:
         """
-        1. Fetch the data from osv api
-        2. unzip the file
-        3. open the file one by one
-        4. yield the json file to parse_advisory_data
+        Yield AdvisoryData using a zipped data dump of OSV data
         """
         url = "https://osv-vulnerabilities.storage.googleapis.com/PyPI/all.zip"
         response = requests.get(url).content
         with ZipFile(BytesIO(response)) as zip_file:
             for file_name in zip_file.namelist():
                 if not file_name.startswith("PYSEC-"):
-                    logger.error(f"NotImplementedError PyPI package file_name: {file_name}")
-                else:
-                    with zip_file.open(file_name) as f:
-                        vul_info = json.load(f)
-                        yield parse_advisory_data(vul_info, supported_ecosystem="pypi")
+                    logger.error(f"Unsupported PyPI advisory data file: {file_name}")
+                    continue
+                with zip_file.open(file_name) as f:
+                    vul_info = json.load(f)
+                    yield parse_advisory_data(raw_data=vul_info, supported_ecosystem="pypi")