Refactor conversion code for pypi to be in shared utilities

Author: ArBridgeman

exasol/toolbox/nox/_dependencies.py

@@ -19,7 +19,10 @@
 import tomlkit
 from nox import Session
 
-from exasol.toolbox.security import GitHubVulnerabilityIssue
+from exasol.toolbox.security import (
+    GitHubVulnerabilityIssue,
+    from_pip_audit,
+)
 
 
 @dataclass(frozen=True)
@@ -388,18 +391,14 @@ def _set_to_resolve(
 
     def _split_resolution_status(self) -> None:
         to_resolve_by_cve = {vuln.cve: vuln for vuln in self.to_resolve}
-        cves_to_resolve = to_resolve_by_cve.keys()
+        cves_to_resolve = set(to_resolve_by_cve.keys())
+
         audit_json, _ = Audit().audit()
+        cve_audit = {vuln.cve for vuln in from_pip_audit(json.dumps(audit_json))}
+
+        cves_not_resolved = cves_to_resolve.intersection(cve_audit)
 
-        not_resolved: list = []
-        for dependency in audit_json["dependencies"]:
-            for v in dependency["vulns"]:
-                vuln_ids = set([v["id"]] + v["aliases"])
-                if remaining_cves := (cves_to_resolve & vuln_ids):
-                    not_resolved.extend(
-                        to_resolve_by_cve[cve] for cve in remaining_cves
-                    )
-        self.not_resolved = set(not_resolved)
+        self.not_resolved = {to_resolve_by_cve[cve] for cve in cves_not_resolved}
         self.resolved = self.to_resolve - self.not_resolved
 
     def get_packages(self) -> set[str]:

exasol/toolbox/security/__init__.py

@@ -7,6 +7,7 @@
     dataclass,
 )
 from enum import Enum
+from typing import Iterable
 
 import typer
 
@@ -105,3 +106,58 @@ def get_link(self, package: str, vuln_id: str) -> str:
             VulnerabilitySource.PYSEC: "https://github.com/pypa/advisory-database/blob/main/vulns/{package}/{vuln_id}.yaml",
         }
         return map_link[self].format(package=package, vuln_id=vuln_id)
+
+
+def from_pip_audit(report: str) -> Iterable[VulnerabilityIssue]:
+    """
+    Transforms the JSON output from `nox -s dependency:audit` into an iterable of
+    `VulnerabilityIssue` objects.
+
+    This does not gracefully handle scenarios where:
+     - a CVE is not initially associated with the vulnerability; however, the assumption
+     is that such vulnerabilities will later be associated with a CVE.
+     - the same vulnerability ID (CVE, PYSEC, GHSA, etc.) is present across
+     multiple coordinates.
+
+    Input:
+        '{"dependencies": [{"name": "<package_name>", "version": "<package_version>",
+        "vulns": [{"id": "<vuln_id>", "fix_versions": ["<fix_version>"],
+        "aliases": ["<vuln_id2>"], "description": "<vuln_description>"}]}]}'
+
+    Args:
+        report:
+            the JSON output of `nox -s dependency:audit` provided as a str
+    """
+    report_dict = json.loads(report)
+    dependencies = report_dict.get("dependencies", [])
+    for dependency in dependencies:
+        package = dependency["name"]
+        for v in dependency["vulns"]:
+            refs = [v["id"]] + v["aliases"]
+            cves, cwes, links = identify_pypi_references(
+                references=refs, package_name=package
+            )
+            if cves:
+                yield VulnerabilityIssue(
+                    cve=sorted(cves)[0],
+                    cwe="None" if not cwes else ", ".join(cwes),
+                    description=v["description"],
+                    coordinates=f"{package}:{dependency['version']}",
+                    references=tuple(links),
+                )
+
+
+def identify_pypi_references(
+    references: list[str], package_name: str
+) -> tuple[list[str], list[str], list[str]]:
+    refs: dict = {k: [] for k in VulnerabilitySource}
+    links = []
+    for reference in references:
+        if source := VulnerabilitySource.from_prefix(reference.upper()):
+            refs[source].append(reference)
+            links.append(source.get_link(package=package_name, vuln_id=reference))
+    return (
+        refs[VulnerabilitySource.CVE],
+        refs[VulnerabilitySource.CWE],
+        links,
+    )

exasol/toolbox/tools/security.py

@@ -21,7 +21,7 @@
 from exasol.toolbox.security import (
     GitHubVulnerabilityIssue,
     VulnerabilityIssue,
-    VulnerabilitySource,
+    from_pip_audit,
 )
 
 stdout = print
@@ -82,61 +82,6 @@ def from_maven(report: str) -> Iterable[VulnerabilityIssue]:
             )
 
 
-def identify_pypi_references(
-    references: list[str], package_name: str
-) -> tuple[list[str], list[str], list[str]]:
-    refs: dict = {k: [] for k in VulnerabilitySource}
-    links = []
-    for reference in references:
-        if source := VulnerabilitySource.from_prefix(reference.upper()):
-            refs[source].append(reference)
-            links.append(source.get_link(package=package_name, vuln_id=reference))
-    return (
-        refs[VulnerabilitySource.CVE],
-        refs[VulnerabilitySource.CWE],
-        links,
-    )
-
-
-def from_pip_audit(report: str) -> Iterable[VulnerabilityIssue]:
-    """
-    Transforms the JSON output from `nox -s dependency:audit` into an iterable of
-    `VulnerabilityIssue` objects.
-
-    This does not gracefully handle scenarios where:
-     - a CVE is not initially associated with the vulnerability; however, the assumption
-     is that such vulnerabilities will later be associated with a CVE.
-     - the same vulnerability ID (CVE, PYSEC, GHSA, etc.) is present across
-     multiple coordinates.
-
-    Input:
-        '{"dependencies": [{"name": "<package_name>", "version": "<package_version>",
-        "vulns": [{"id": "<vuln_id>", "fix_versions": ["<fix_version>"],
-        "aliases": ["<vuln_id2>"], "description": "<vuln_description>"}]}]}'
-
-    Args:
-        report:
-            the JSON output of `nox -s dependency:audit` provided as a str
-    """
-    report_dict = json.loads(report)
-    dependencies = report_dict.get("dependencies", [])
-    for dependency in dependencies:
-        package = dependency["name"]
-        for v in dependency["vulns"]:
-            refs = [v["id"]] + v["aliases"]
-            cves, cwes, links = identify_pypi_references(
-                references=refs, package_name=package
-            )
-            if cves:
-                yield VulnerabilityIssue(
-                    cve=sorted(cves)[0],
-                    cwe="None" if not cwes else ", ".join(cwes),
-                    description=v["description"],
-                    coordinates=f"{package}:{dependency['version']}",
-                    references=tuple(links),
-                )
-
-
 @dataclass(frozen=True)
 class SecurityIssue:
     file_name: str

test/unit/security_test.py

@@ -11,6 +11,8 @@
 from exasol.toolbox.security import (
     GitHubVulnerabilityIssue,
     VulnerabilitySource,
+    from_pip_audit,
+    identify_pypi_references,
 )
 from exasol.toolbox.tools import security
 
@@ -545,14 +547,14 @@ def test_from_prefix(prefix: str, expected):
     ],
 )
 def test_identify_pypi_references(reference: str, expected):
-    actual = security.identify_pypi_references([reference], package_name="dummy")
+    actual = identify_pypi_references([reference], package_name="dummy")
     assert actual == expected
 
 
 class TestFromPipAudit:
     @staticmethod
     def test_no_vulnerability_returns_empty_list():
-        actual = set(security.from_pip_audit("{}"))
+        actual = set(from_pip_audit("{}"))
         assert actual == set()
 
     @staticmethod
@@ -562,5 +564,5 @@ def test_convert_vulnerability_to_issue(
         audit_json = json.dumps(pip_audit_report)
         expected = {pip_audit_jinja2_issue, pip_audit_cryptography_issue}
 
-        actual = set(security.from_pip_audit(audit_json))
+        actual = set(from_pip_audit(audit_json))
         assert actual == expected