Streamline code

Use variables when relevant

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

Author: pombredanne

vulntotal/datasources/deps.py

@@ -57,21 +57,22 @@ def supported_ecosystem(cls):
 
 
 def parse_advisory(advisory) -> Iterable[VendorData]:
-    affected_versions = [event["version"] for event in advisory["packages"][0]["versionsAffected"]]
-    fixed_versions = [event["version"] for event in advisory["packages"][0]["versionsUnaffected"]]
+    package = advisory["packages"][0]
+    affected_versions = [event["version"] for event in package["versionsAffected"]]
+    fixed_versions = [event["version"] for event in package["versionsUnaffected"]]
     yield VendorData(
-        aliases=sorted(list(set(advisory["aliases"]))),
-        affected_versions=sorted(list(set(affected_versions))),
-        fixed_versions=sorted(list(set(fixed_versions))),
+        aliases=sorted(set(advisory["aliases"])),
+        affected_versions=sorted(set(affected_versions)),
+        fixed_versions=sorted(set(fixed_versions)),
     )
 
 
 def parse_advisories_from_meta(advisories_metadata):
     advisories = []
-    if "dependencies" in advisories_metadata and advisories_metadata["dependencies"]:
-        for dependency in advisories_metadata["dependencies"]:
-            if dependency["advisories"]:
-                advisories.extend(dependency["advisories"])
+    dependencies = advisories_metadata.get("dependencies") or []
+    for dependency in dependencies:
+        advs = dependency.get("advisories") or []
+        advisories.extend(advs)
     return advisories
 
 

vulntotal/datasources/github.py

@@ -56,11 +56,12 @@ def supported_ecosystem(cls):
 
 def parse_advisory(interesting_edges) -> Iterable[VendorData]:
     for edge in interesting_edges:
-        aliases = [aliase["value"] for aliase in edge["node"]["advisory"]["identifiers"]]
+        node = edge["node"]
+        aliases = [aliase["value"] for aliase in node["advisory"]["identifiers"]]
         affected_versions = (
-            edge["node"]["vulnerableVersionRange"].strip().replace(" ", "").split(",")
+            node["vulnerableVersionRange"].strip().replace(" ", "").split(",")
         )
-        fixed_versions = [edge["node"]["firstPatchedVersion"]["identifier"]]
+        fixed_versions = [node["firstPatchedVersion"]["identifier"]]
         yield VendorData(
             aliases=sorted(list(set(aliases))),
             affected_versions=sorted(list(set(affected_versions))),

vulntotal/datasources/gitlab.py

@@ -139,12 +139,13 @@ def get_casesensitive_slug(path, package_slug):
         paginated_tree = response[0]["data"]["project"]["repository"]["paginatedTree"]
 
         for slug in paginated_tree["nodes"][0]["trees"]["nodes"]:
-            if slug["flatPath"].lower() == package_slug.lower():
-                return slug["flatPath"]
+            slug_flatpath = slug["flatPath"]
+            if slug_flatpath.lower() == package_slug.lower():
+                return slug_flatpath
 
             # If the namespace/subfolder contains multiple packages, then progressive transverse through folders tree
-            if package_slug.lower().startswith(slug["flatPath"].lower()):
-                return get_gitlab_style_slug(slug["flatPath"], package_slug)
+            if package_slug.lower().startswith(slug_flatpath.lower()):
+                return get_gitlab_style_slug(slug_flatpath, package_slug)
 
         payload[0]["variables"]["nextPageCursor"] = paginated_tree["pageInfo"]["endCursor"]
         hasnext = paginated_tree["pageInfo"]["hasNextPage"]
@@ -157,10 +158,11 @@ def parse_interesting_advisories(location, version, delete_download=False) -> It
     for file in sorted(files):
         with open(file) as f:
             gitlab_advisory = saneyaml.load(f)
-        if gitlab_constraints_satisfied(gitlab_advisory["affected_range"], version):
+        affected_range = gitlab_advisory["affected_range"]
+        if gitlab_constraints_satisfied(affected_range, version):
             yield VendorData(
                 aliases=gitlab_advisory["identifiers"],
-                affected_versions=[gitlab_advisory["affected_range"]],
+                affected_versions=[affected_range],
                 fixed_versions=gitlab_advisory["fixed_versions"],
             )
     if delete_download:

vulntotal/datasources/oss_index.py

@@ -81,15 +81,15 @@ def supported_ecosystem(cls):
 
 def parse_advisory(component) -> Iterable[VendorData]:
     response = component[0]
-    if response["vulnerabilities"]:
-        for vuln in response["vulnerabilities"]:
-            aliases = [vuln["id"]]
-            affected_versions = []
-            fixed_versions = []
-            if "versionRanges" in vuln:
-                affected_versions.extend(vuln["versionRanges"])
-            yield VendorData(
-                aliases=aliases,
-                affected_versions=affected_versions,
-                fixed_versions=fixed_versions,
-            )
+    vulnerabilities = response.get("vulnerabilities") or []
+    for vuln in vulnerabilities:
+        aliases = [vuln["id"]]
+        affected_versions = []
+        fixed_versions = []
+        version_ranges = vuln.get("versionRanges") or []
+        affected_versions.extend(version_ranges)
+        yield VendorData(
+            aliases=aliases,
+            affected_versions=affected_versions,
+            fixed_versions=fixed_versions,
+        )

vulntotal/datasources/osv.py

@@ -63,15 +63,18 @@ def supported_ecosystem(cls):
 
 
 def parse_advisory(response) -> Iterable[VendorData]:
-    """Parse response from OSV API and yield VendorData"""
+    """
+    Parse response from OSV API and yield VendorData
+    """
 
     for vuln in response.get("vulns") or []:
         aliases = []
         affected_versions = []
         fixed = []
 
         aliases.extend(vuln.get("aliases") or [])
-        aliases.append(vuln.get("id")) if vuln.get("id") else None
+        vuln_id = vuln.get("id")
+        aliases.append(vuln_id) if vuln_id else None
 
         try:
             affected_versions.extend(get_item(vuln, "affected", 0, "versions") or [])
@@ -100,40 +103,43 @@ def generate_payload(purl):
     supported_ecosystem = OSVDataSource.supported_ecosystem()
     payload = {}
     payload["version"] = purl.version
-    payload["package"] = {}
+    package = payload["package"] = {}
 
-    if purl.type in supported_ecosystem:
-        payload["package"]["ecosystem"] = supported_ecosystem[purl.type]
+    purl_type = purl.type
+    purl_namespace = purl.namespace
 
-    if purl.type == "maven":
-        if not purl.namespace:
+    if purl_type in supported_ecosystem:
+        package["ecosystem"] = supported_ecosystem[purl_type]
+
+    if purl_type == "maven":
+        if not purl_namespace:
             logger.error(f"Invalid Maven PURL {str(purl)}")
             return
-        payload["package"]["name"] = f"{purl.namespace}:{purl.name}"
+        package["name"] = f"{purl.namespace}:{purl.name}"
 
-    elif purl.type == "packagist":
-        if not purl.namespace:
+    elif purl_type == "packagist":
+        if not purl_namespace:
             logger.error(f"Invalid Packagist PURL {str(purl)}")
             return
-        payload["package"]["name"] = f"{purl.namespace}/{purl.name}"
+        package["name"] = f"{purl.namespace}/{purl.name}"
 
-    elif purl.type == "linux":
+    elif purl_type == "linux":
         if purl.name not in ("kernel", "Kernel"):
             logger.error(f"Invalid Linux PURL {str(purl)}")
             return
-        payload["package"]["name"] = "Kernel"
+        package["name"] = "Kernel"
 
-    elif purl.type == "nuget":
+    elif purl_type == "nuget":
         nuget_package = get_closest_nuget_package_name(purl.name)
         if not nuget_package:
             logger.error(f"Invalid NuGet PURL {str(purl)}")
             return
-        payload["package"]["name"] = nuget_package
+        package["name"] = nuget_package
 
-    elif purl.type == "golang" and purl.namespace:
-        payload["package"]["name"] = f"{purl.namespace}/{purl.name}"
+    elif purl_type == "golang" and purl_namespace:
+        package["name"] = f"{purl.namespace}/{purl.name}"
 
     else:
-        payload["package"]["name"] = purl.name
+        package["name"] = purl.name
 
     return payload