Add JSON output for pip-audit for security pipeline

Author: ArBridgeman

doc/changes/unreleased.md

@@ -5,3 +5,4 @@
 * [#73](https://github.com/exasol/python-toolbox/issues/73): Added nox target for auditing work spaces in regard to known vulnerabilities
 * [#65](https://github.com/exasol/python-toolbox/issues/65): Added a Nox task for checking if the changelog got updated.
 * [#369](https://github.com/exasol/python-toolbox/issues/369): Removed option `-v` for `isort`
+* [#372](https://github.com/exasol/python-toolbox/issues/372): Added conversion from audited dependencies json to expected GitHub Issue format 

exasol/toolbox/nox/_dependencies.py

@@ -1,11 +1,18 @@
 from __future__ import annotations
 
+import argparse
+import json
 import subprocess
 import tempfile
 from dataclasses import dataclass
+from enum import (
+    Enum,
+    auto,
+)
 from inspect import cleandoc
 from json import loads
 from pathlib import Path
+from typing import Tuple
 
 import nox
 import tomlkit
@@ -211,8 +218,87 @@ def _normalize_package_name(name: str) -> str:
     return template.format(heading=heading(), rows=rows)
 
 
-def _audit(session: Session) -> None:
-    session.run("poetry", "run", "pip-audit")
+class PipAuditFormat(Enum):
+    columns = auto()
+    json = auto()
+
+    @classmethod
+    def _missing_(cls, value):
+        if isinstance(value, str):
+            for member in cls:
+                if member.name == value.lower():
+                    return member
+        return None
+
+    @classmethod
+    def name_tuple(cls) -> Tuple:
+        return tuple(fmt.name for fmt in PipAuditFormat)
+
+
+class Audit:
+    @staticmethod
+    def _filter_json_for_vulnerabilities(audit_json_bytes: bytes) -> dict:
+        """filters json for only packages with vulnerabilities"""
+        audit_dict = json.loads(audit_json_bytes.decode("utf-8"))
+        return {
+            "dependencies": [
+                {
+                    "name": entry["name"],
+                    "version": entry["version"],
+                    "vulns": entry["vulns"],
+                }
+                for entry in audit_dict["dependencies"]
+                if entry["vulns"]
+            ]
+        }
+
+    @staticmethod
+    def _parse_format(session) -> argparse.Namespace:
+        parser = argparse.ArgumentParser(
+            description="Audits dependencies for security vulnerabilities",
+            usage="nox -s dependency:audit -- -- [options]",
+        )
+        parser.add_argument(
+            "-f",
+            "--format",
+            type=str,
+            default=PipAuditFormat.columns.name,
+            help="Format to emit audit results in",
+            choices=PipAuditFormat.name_tuple(),
+        )
+        parser.add_argument(
+            "-o",
+            "--output",
+            type=Path,
+            default=None,
+            help="Output results to the given file",
+        )
+        return parser.parse_args(args=session.posargs)
+
+    def run(self, session: Session) -> None:
+        args = self._parse_format(session)
+        audit_format = PipAuditFormat[args.format]
+
+        command = ["poetry", "run", "pip-audit", "-f", audit_format.name]
+        if audit_format == PipAuditFormat.columns:
+            if args.output:
+                command.extend(["-o", args.output])
+            session.run(*command)
+
+        elif audit_format == PipAuditFormat.json:
+            output = subprocess.run(command, capture_output=True)
+            audit_json = self._filter_json_for_vulnerabilities(output.stdout)
+
+            if args.output:
+                with open(args.output, "w") as file:
+                    json.dump(audit_json, file)
+            else:
+                print(audit_json)
+
+            if output.returncode != 0:
+                session.warn(
+                    f"Command {' '.join(command)} failed with exit code {output.returncode}",
+                )
 
 
 @nox.session(name="dependency:licenses", python=False)
@@ -227,4 +313,4 @@ def dependency_licenses(session: Session) -> None:
 @nox.session(name="dependency:audit", python=False)
 def audit(session: Session) -> None:
     """Check for known vulnerabilities"""
-    _audit(session)
+    Audit().run(session=session)

exasol/toolbox/tools/security.py

@@ -16,7 +16,10 @@
 from functools import partial
 from inspect import cleandoc
 from pathlib import Path
-from typing import Tuple
+from typing import (
+    List,
+    Tuple,
+)
 
 import typer
 
@@ -102,6 +105,59 @@ def from_maven(report: str) -> Iterable[Issue]:
             )
 
 
+class VulnerabilitySource(str, Enum):
+    CVE = "CVE"
+    CWE = "CWE"
+    GHSA = "GHSA"
+    PYSEC = "PYSEC"
+
+    def get_link(self, package: str, vuln_id: str) -> str:
+        if self.value == VulnerabilitySource.CVE:
+            return f"https://nvd.nist.gov/vuln/detail/{vuln_id}"
+        if self.value == VulnerabilitySource.CWE:
+            cwe_id = vuln_id.upper().replace(f"{VulnerabilitySource.CWE}-", "")
+            return f"https://cwe.mitre.org/data/definitions/{cwe_id}.html"
+        if self.value == VulnerabilitySource.GHSA:
+            return f"https://github.com/advisories/{vuln_id}"
+        return f"https://github.com/pypa/advisory-database/blob/main/vulns/{package}/{vuln_id}.yaml"
+
+
+def identify_pypi_references(references: List[str], package_name: str):
+    ref_cves, ref_cwes, ref_links = [], [], []
+    for reference in references:
+        for source in VulnerabilitySource:
+            if reference.upper().startswith(source.value):
+                if source == VulnerabilitySource.CVE:
+                    ref_cves.append(reference)
+                elif source == VulnerabilitySource.CWE:
+                    ref_cwes.append(reference)
+                ref_links.append(
+                    source.get_link(package=package_name, vuln_id=reference)
+                )
+                continue
+    return ref_cves, ref_cwes, ref_links
+
+
+def from_python(report: str) -> Iterable[Issue]:
+    # Note: Consider adding warnings if there is the same cve with multiple coordinates
+    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
+            )
+            yield Issue(
+                cve="None" if not cves else cves[0],
+                cwe="None" if not cwes else cwes[0],
+                description=v["description"],
+                coordinates=f"{package}:{dependency['version']}",
+                references=tuple(links),
+            )
+
+
 @dataclass(frozen=True)
 class SecurityIssue:
     file_name: str
@@ -220,6 +276,7 @@ def create_security_issue(issue: Issue, project="") -> tuple[str, str]:
 
 class Format(str, Enum):
     Maven = "maven"
+    Python = "python"
 
 
 # pylint: disable=redefined-builtin
@@ -243,7 +300,13 @@ def _maven(infile):
             stdout(issue)
         raise typer.Exit(code=0)
 
-    actions = {Format.Maven: _maven}
+    def _python(infile):
+        issues = from_python(infile.read())
+        for issue in _issues_as_json_str(issues):
+            stdout(issue)
+        raise typer.Exit(code=0)
+
+    actions = {Format.Maven: _maven, Format.Python: _python}
     action = actions[format]
     action(input_file)
 

test/unit/conftest.py

@@ -0,0 +1,55 @@
+from inspect import cleandoc
+
+import pytest
+from exasol.toolbox.tools import security
+
+
+@pytest.fixture(scope="session")
+def pip_audit_issue():
+    return security.Issue(
+        cve="CVE-2025-27516",
+        cwe="None",
+        description=cleandoc(
+            """An oversight in how the Jinja sandboxed environment
+              interacts with the `|attr` filter allows an attacker
+              that controls the content of a template to execute
+              arbitrary Python code.  To exploit the vulnerability,
+              an attacker needs to control the content of a template.
+              Whether that is the case depends on the type of
+              application using Jinja. This vulnerability impacts
+              users of applications which execute untrusted templates.
+              Jinja's sandbox does catch calls to `str.format` and
+              ensures they don't escape the sandbox. However, it's
+              possible to use the `|attr` filter to get a reference to
+              a string's plain format method, bypassing the sandbox.
+              After the fix, the `|attr` filter no longer bypasses
+              the environment's attribute lookup.
+              """
+        ),
+        coordinates="jinja2:3.1.5",
+        references=(
+            "https://github.com/advisories/GHSA-cpwx-vrp4-4pq7",
+            "https://nvd.nist.gov/vuln/detail/CVE-2025-27516",
+        ),
+    )
+
+
+@pytest.fixture(scope="session")
+def pip_audit_report(pip_audit_issue):
+    name, version = pip_audit_issue.coordinates.split(":")
+    return {
+        "dependencies": [
+            {
+                "name": name,
+                "version": version,
+                "vulns": [
+                    {
+                        "id": "GHSA-cpwx-vrp4-4pq7",
+                        "fix_versions": ["3.1.6"],
+                        "aliases": [pip_audit_issue.cve],
+                        "description": pip_audit_issue.description,
+                    }
+                ],
+            }
+        ]
+    }

test/unit/dependencies_test.py

@@ -1,6 +1,9 @@
+import json
+
 import pytest
 
 from exasol.toolbox.nox._dependencies import (
+    Audit,
     Package,
     _dependencies,
     _normalize,
@@ -157,3 +160,24 @@ def test_packages_from_json(json, expected):
 def test_packages_to_markdown(dependencies, packages, expected):
     actual = _packages_to_markdown(dependencies, packages)
     assert actual == expected
+
+
+class TestFilterJsonForVulnerabilities:
+
+    @staticmethod
+    def test_no_vulnerability_returns_empty_list():
+        audit_dict = {
+            "dependencies": [{"name": "alabaster", "version": "0.7.16", "vulns": []}]
+        }
+        audit_json = json.dumps(audit_dict).encode("utf-8")
+        expected = {"dependencies": []}
+
+        actual = Audit._filter_json_for_vulnerabilities(audit_json)
+        assert actual == expected
+
+    @staticmethod
+    def test_vulnerability_returned_in_list(pip_audit_report):
+        audit_json = json.dumps(pip_audit_report).encode("utf-8")
+
+        actual = Audit._filter_json_for_vulnerabilities(audit_json)
+        assert actual == pip_audit_report

test/unit/security_test.py

@@ -462,3 +462,62 @@ def test_from_json(json_file, expected):
         references=expected["references"],
     )
     assert list(actual) == [expected_issue]
+
+
+@pytest.mark.parametrize(
+    "reference, expected",
+    [
+        pytest.param(
+            "CVE-2025-27516",
+            (
+                ["CVE-2025-27516"],
+                [],
+                ["https://nvd.nist.gov/vuln/detail/CVE-2025-27516"],
+            ),
+            id="CVE_identified_with_link",
+        ),
+        pytest.param(
+            "CWE-611",
+            ([], ["CWE-611"], ["https://cwe.mitre.org/data/definitions/611.html"]),
+            id="CWE_identified_with_link",
+        ),
+        pytest.param(
+            "GHSA-cpwx-vrp4-4pq7",
+            ([], [], ["https://github.com/advisories/GHSA-cpwx-vrp4-4pq7"]),
+            id="GHSA_link",
+        ),
+        pytest.param(
+            "PYSEC-2025-9",
+            (
+                [],
+                [],
+                [
+                    "https://github.com/pypa/advisory-database/blob/main/vulns/dummy/PYSEC-2025-9.yaml"
+                ],
+            ),
+            id="PYSEC_link",
+        ),
+    ],
+)
+def test_identify_pypi_references(reference: str, expected):
+    actual = security.identify_pypi_references([reference], package_name="dummy")
+    assert actual == expected
+
+
+class TestFromPython:
+    @staticmethod
+    def test_no_vulnerability_returns_empty_list():
+        actual = set(security.from_python("{}"))
+        assert actual == set()
+
+    @staticmethod
+    def test_convert_vulnerability_to_issue(pip_audit_report, pip_audit_issue):
+        audit_json = json.dumps(pip_audit_report)
+
+        actual = set(security.from_python(audit_json)).pop()
+
+        assert actual.cve == pip_audit_issue.cve
+        assert actual.cwe == pip_audit_issue.cwe
+        assert actual.description == pip_audit_issue.description
+        assert actual.coordinates == pip_audit_issue.coordinates
+        assert actual.references == pip_audit_issue.references