Add nox task dependency:update for workflows

Author: ArBridgeman

doc/changes/unreleased.md

@@ -2,10 +2,11 @@
 
 ## ✨ Features
 
-* [#73](https://github.com/exasol/python-toolbox/issues/73): Added nox target for auditing work spaces in regard to known vulnerabilities
+* [#73](https://github.com/exasol/python-toolbox/issues/73): Added Nox task 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 pip-audit JSON to expected GitHub Issue format
+* [#382](https://github.com/exasol/python-toolbox/issues/382) Added Nox task to update vulnerable dependencies
 
 ## ⚒️ Refactorings
 * [#388](https://github.com/exasol/python-toolbox/issues/388): Switch GitHub workflows to use pinned OS version

exasol/toolbox/nox/_dependencies.py

@@ -2,17 +2,25 @@
 
 import argparse
 import json
+import re
 import subprocess
 import tempfile
-from dataclasses import dataclass
+from contextlib import contextmanager
+from dataclasses import (
+    dataclass,
+    field,
+)
 from inspect import cleandoc
 from json import loads
 from pathlib import Path
+from subprocess import CompletedProcess
 
 import nox
 import tomlkit
 from nox import Session
 
+from exasol.toolbox.security import GitHubVulnerabilityIssue
+
 
 @dataclass(frozen=True)
 class Package:
@@ -256,13 +264,15 @@ def _parse_args(session) -> argparse.Namespace:
         )
         return parser.parse_args(args=session.posargs)
 
-    def run(self, session: Session) -> None:
-        args = self._parse_args(session)
-
-        command = ["poetry", "run", "pip-audit", "-f", "json"]
+    def audit(self) -> tuple[dict, CompletedProcess]:
+        command = ("poetry", "run", "pip-audit", "-f", "json")
         output = subprocess.run(command, capture_output=True)
-
         audit_json = self._filter_json_for_vulnerabilities(output.stdout)
+        return audit_json, output
+
+    def run(self, session: Session) -> None:
+        args = self._parse_args(session)
+        audit_json, output = self.audit()
         if args.output:
             with open(args.output, "w") as file:
                 json.dump(audit_json, file)
@@ -271,8 +281,230 @@ def run(self, session: Session) -> None:
 
         if output.returncode != 0:
             session.warn(
-                f"Command {' '.join(command)} failed with exit code {output.returncode}",
+                f"Command {' '.join(output.args)} failed with exit code {output.returncode}",
+            )
+
+
+@dataclass(frozen=True)
+class PackageVersion:
+    name: str
+    version: str
+
+
+@dataclass
+class PackageVersionTracker:
+    """
+    Tracks direct dependencies for package versions before & after updates
+
+    Assumption:
+      - The dependency ranges in the pyproject.toml allows users to often update
+      transitive dependencies on their own. It is, therefore, more important for us to
+      track the changes of direct dependencies and, if present, the resolution of both
+      vulnerabilities for direct and transitive dependencies.
+    """
+
+    before_env: set[PackageVersion] = field(default_factory=set)
+    after_env: set[PackageVersion] = field(default_factory=set)
+
+    @staticmethod
+    def _obtain_version_set() -> set[PackageVersion]:
+        def _get_package_version(line: str) -> PackageVersion:
+            pattern = r"\s+(\d+(?:\.\d+)*)\s+"
+            groups = re.split(pattern, line)
+            return PackageVersion(name=groups[0], version=groups[1])
+
+        command = ("poetry", "show", "--top-level")
+        result = subprocess.run(command, capture_output=True, check=True)
+        return {
+            _get_package_version(line)
+            for line in result.stdout.decode("utf-8").splitlines()
+        }
+
+    @property
+    def changes(self) -> tuple:
+        before_update_dict = {pkg.name: pkg for pkg in self.before_env}
+        after_update_dict = {pkg.name: pkg for pkg in self.after_env}
+
+        def _get_change_str(pkg_name: str) -> str | None:
+            if pkg_name not in after_update_dict.keys():
+                entry = before_update_dict[pkg_name]
+                return f"* Removed {entry.name} ({entry.version})"
+            if pkg_name not in before_update_dict.keys():
+                entry = after_update_dict[pkg_name]
+                return f"* Added {entry.name} ({entry.version})"
+            before_entry = before_update_dict[pkg_name]
+            after_entry = after_update_dict[pkg_name]
+            if before_entry.version != after_entry.version:
+                return f"* Updated {pkg_name} ({before_entry.version} → {after_entry.version})"
+            return None
+
+        all_packages = before_update_dict.keys() | after_update_dict.keys()
+        return tuple(
+            change_str
+            for pkg_name in all_packages
+            if (change_str := _get_change_str(pkg_name))
+        )
+
+    @property
+    def packages(self) -> set[str]:
+        return {pkg.name for pkg in self.before_env}
+
+    def __enter__(self) -> PackageVersionTracker:
+        self.before_env = self._obtain_version_set()
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self.after_env = self._obtain_version_set()
+
+
+@contextmanager
+def managed_file(file_obj: argparse.FileType):
+    """Context manager to manage a file provided by argparse"""
+    yield file_obj
+
+
+@dataclass
+class VulnerabilityTracker:
+    """Tracks the resolution of GitHubVulnerabilityIssues before & after updates"""
+
+    to_resolve: set[GitHubVulnerabilityIssue] = field(default_factory=set)
+    resolved: set[GitHubVulnerabilityIssue] = field(default_factory=set)
+    not_resolved: set[GitHubVulnerabilityIssue] = field(default_factory=set)
+
+    def __init__(self, vulnerability_issues: argparse.FileType | None):
+        self.to_resolve: set[GitHubVulnerabilityIssue] = self._set_to_resolve(
+            vulnerability_issues
+        )
+
+    @staticmethod
+    def _set_to_resolve(
+        vulnerability_issues: argparse.FileType | None,
+    ) -> set[GitHubVulnerabilityIssue]:
+        if not vulnerability_issues:
+            return set()
+        with managed_file(vulnerability_issues) as f:
+            lines = f.readlines()
+        return set(GitHubVulnerabilityIssue.extract_from_jsonl(lines))
+
+    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()
+        audit_json, _ = Audit().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.resolved = self.to_resolve - self.not_resolved
+
+    def get_packages(self) -> set[str]:
+        return {vuln.coordinates.split(":")[0] for vuln in self.to_resolve}
+
+    @property
+    def issues_not_resolved(self) -> tuple[str, ...]:
+        return tuple(
+            f"* Did NOT resolve {vuln.issue_url} ({vuln.cve})"
+            for vuln in self.not_resolved
+        )
+
+    @property
+    def issues_resolved(self) -> tuple[str, ...]:
+        return tuple(
+            f"* Closes {vuln.issue_url} ({vuln.cve})" for vuln in self.resolved
+        )
+
+    @property
+    def summary(self) -> tuple[str, ...]:
+        return tuple(
+            cleandoc(
+                f"""{vuln.cve} in dependency `{vuln.coordinates}`\n {vuln.description}
+            """
             )
+            for vuln in self.resolved
+        )
+
+    @property
+    def vulnerabilities_resolved(self) -> tuple[str, ...]:
+        def get_issue_number(issue_url: str) -> str | None:
+            pattern = r"/issues/(\d+)$"
+            match = re.search(pattern, issue_url)
+            return match.group(1) if match else None
+
+        return tuple(
+            f"* #{get_issue_number(vuln.issue_url)} Fixed vulnerability {vuln.cve} in `{vuln.coordinates}`"
+            for vuln in self.resolved
+        )
+
+    def __enter__(self) -> VulnerabilityTracker:
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        self._split_resolution_status()
+
+
+@dataclass(frozen=True)
+class DependencyChanges:
+    package_changes: tuple[str, ...]
+    issues_resolved: tuple[str, ...]
+    issues_not_resolved: tuple[str, ...]
+    vulnerabilities_resolved: tuple[str, ...]
+    vulnerabilities_resolved_summary: tuple[str, ...]
+
+
+class DependencyUpdate:
+    """Update dependencies"""
+
+    @staticmethod
+    def _parse_args(session) -> argparse.Namespace:
+        parser = argparse.ArgumentParser(
+            description="Updates dependencies & returns changes",
+            usage="nox -s dependency:audit -- -- [options]",
+        )
+        parser.add_argument(
+            "-v",
+            "--vulnerability-issues",
+            type=argparse.FileType("r"),
+            default=None,
+            help="JSONL of vulnerabilities (of type `GitHubVulnerabilityIssue`)",
+        )
+        return parser.parse_args(args=session.posargs)
+
+    @staticmethod
+    def _perform_basic_vulnerability_update(
+        pkg_tracker: PackageVersionTracker, vuln_tracker: VulnerabilityTracker
+    ) -> None:
+        vuln_packages = vuln_tracker.get_packages()
+
+        # vulnerabilities of direct dependencies require a pyproject.toml update
+        vuln_direct_dependencies = vuln_packages.intersection(pkg_tracker.packages)
+        if vuln_direct_dependencies:
+            command = ("poetry", "up") + tuple(vuln_direct_dependencies)
+            subprocess.run(command, capture_output=True)
+
+        command = ("poetry", "update") + tuple(vuln_packages)
+        subprocess.run(command, capture_output=True)
+
+    def run(self, session: Session) -> DependencyChanges:
+        """Update the dependencies associated with GitHubVulnerabilityIssues"""
+        args = self._parse_args(session)
+        with PackageVersionTracker() as pkg_tracker:
+            with VulnerabilityTracker(args.vulnerability_issues) as vuln_tracker:
+                self._perform_basic_vulnerability_update(
+                    pkg_tracker=pkg_tracker, vuln_tracker=vuln_tracker
+                )
+
+        return DependencyChanges(
+            package_changes=pkg_tracker.changes,
+            issues_resolved=vuln_tracker.issues_resolved,
+            issues_not_resolved=vuln_tracker.issues_not_resolved,
+            vulnerabilities_resolved=vuln_tracker.vulnerabilities_resolved,
+            vulnerabilities_resolved_summary=vuln_tracker.summary,
+        )
 
 
 @nox.session(name="dependency:licenses", python=False)
@@ -288,3 +520,19 @@ def dependency_licenses(session: Session) -> None:
 def audit(session: Session) -> None:
     """Check for known vulnerabilities"""
     Audit().run(session=session)
+
+
+@nox.session(name="dependency:update", python=False)
+def update(session: Session) -> None:
+    """Updates dependencies & returns changes"""
+    dependency_changes = DependencyUpdate().run(session)
+    print("Resolved issues")
+    print(*dependency_changes.issues_resolved, sep="\n")
+    print("\nNot resolved issues")
+    print(*dependency_changes.issues_not_resolved, sep="\n")
+    print("\nSummary")
+    print(*dependency_changes.vulnerabilities_resolved_summary, sep="\n")
+    print("\nSecurity fixes")
+    print(*dependency_changes.vulnerabilities_resolved, sep="\n")
+    print("\nDependencies")
+    print(*dependency_changes.package_changes, sep="\n")

test/unit/conftest.py

@@ -54,6 +54,14 @@ def pip_audit_cryptography_issue():
     )
 
 
+@pytest.fixture(scope="session")
+def pip_audit_cryptography_github_issue(pip_audit_cryptography_issue):
+    return security.GitHubVulnerabilityIssue.from_vulnerability_issue(
+        issue=pip_audit_cryptography_issue,
+        issue_url="https://github.com/exasol/<repo-name>/issues/394",
+    )
+
+
 @pytest.fixture(scope="session")
 def pip_audit_report(pip_audit_jinja2_issue, pip_audit_cryptography_issue):
     jinja2_name, jinja2_version = pip_audit_jinja2_issue.coordinates.split(":")