Refactor VulnerabilityIssue class & functions to be more explicit & with docstrings

- Alter & add missing unit tests

Author: ArBridgeman

doc/github_actions/security_issues.rst

@@ -106,7 +106,7 @@ Ideas
 
 .. todo::
 
-    Add additional details to the :code:`security.Issue` type
+    Add additional details to the :code:`VulnerabilityIssue` type
 
 
 .. todo::

exasol/toolbox/security/__init__.py

@@ -0,0 +1,80 @@
+from __future__ import annotations
+
+import json
+from collections.abc import Generator
+from dataclasses import (
+    asdict,
+    dataclass,
+)
+
+import typer
+
+
+@dataclass(frozen=True)
+class VulnerabilityIssue:
+    """
+    Dataclass for a vulnerability to be submitted to GitHub as an issue.
+
+    This format does not reflect the official CVE JSON schema:
+    https://github.com/CVEProject/cve-schema/blob/master/schema/v5.0/CVE_JSON_5.0_schema.json
+    Additionally, it is a known that some vulnerabilities may not initially or ever be
+    assigned a CVE, which is meant to act as a unique identifier. In such cases, they
+    instead have another kind of vulnerability ID associated with them.
+    """
+
+    cve: str
+    cwe: str
+    description: str
+    coordinates: str
+    references: tuple
+
+    @staticmethod
+    def extract_from_jsonl(jsonl: typer.FileText) -> Generator[VulnerabilityIssue]:
+        """Converts the lines of a JSONL file into a generator of VulnerabilityIssue"""
+        lines = (l for l in jsonl if l.strip() != "")
+        for line in lines:
+            obj = json.loads(line)
+            obj["references"] = tuple(obj["references"])
+            yield VulnerabilityIssue(**obj)
+
+    @property
+    def json_str(self) -> str:
+        """Converts to a string-encoded JSON"""
+        issue = asdict(self)
+        return json.dumps(issue)
+
+
+@dataclass(frozen=True)
+class GitHubVulnerabilityIssue:
+    """Dataclass for an existing GitHub Issue associated with a vulnerability."""
+
+    cve: str
+    cwe: str
+    description: str
+    coordinates: str
+    references: tuple
+    issue_url: str
+
+    @staticmethod
+    def from_vulnerability_issue(
+        issue: VulnerabilityIssue, issue_url: str
+    ) -> GitHubVulnerabilityIssue:
+        """Converts VulnerabilityIssue to GitHubVulnerabilityIssue"""
+        return GitHubVulnerabilityIssue(**asdict(issue), issue_url=issue_url.strip())
+
+    @staticmethod
+    def extract_from_jsonl(
+        jsonl: typer.FileText,
+    ) -> Generator[GitHubVulnerabilityIssue]:
+        """Converts the lines of a JSONL file into a generator of GitHubVulnerabilityIssue"""
+        lines = (l for l in jsonl if l.strip() != "")
+        for line in lines:
+            obj = json.loads(line)
+            obj["references"] = tuple(obj["references"])
+            yield GitHubVulnerabilityIssue(**obj)
+
+    @property
+    def json_str(self) -> str:
+        """Converts to a string-encoded JSON"""
+        issue_json = asdict(self)
+        return json.dumps(issue_json)

exasol/toolbox/tools/security.py

@@ -10,45 +10,23 @@
     Generator,
     Iterable,
 )
-from dataclasses import (
-    asdict,
-    dataclass,
-)
+from dataclasses import dataclass
 from enum import Enum
 from functools import partial
 from inspect import cleandoc
 from pathlib import Path
 
 import typer
 
+from exasol.toolbox.security import (
+    GitHubVulnerabilityIssue,
+    VulnerabilityIssue,
+)
+
 stdout = print
 stderr = partial(print, file=sys.stderr)
 
 
-# Note: In the long term we may want to adapt the official CVE json schema
-# https://github.com/CVEProject/cve-schema/blob/master/schema/v5.0/CVE_JSON_5.0_schema.json
-@dataclass(frozen=True)
-class Issue:
-    cve: str
-    cwe: str
-    description: str
-    coordinates: str
-    references: tuple
-
-
-def _issues(input) -> Generator[Issue]:
-    lines = (l for l in input if l.strip() != "")
-    for line in lines:
-        obj = json.loads(line)
-        yield Issue(**obj)
-
-
-def _issues_as_json_str(issues):
-    for issue in issues:
-        issue = asdict(issue)
-        yield json.dumps(issue)
-
-
 def gh_security_issues() -> Generator[tuple[str, str]]:
     """
     Yields issue-id, cve-id pairs for all (closed, open) issues associated with CVEs
@@ -87,14 +65,14 @@ def gh_security_issues() -> Generator[tuple[str, str]]:
     return issues
 
 
-def from_maven(report: str) -> Iterable[Issue]:
+def from_maven(report: str) -> Iterable[VulnerabilityIssue]:
     # Note: Consider adding warnings if there is the same cve with multiple coordinates
     report = json.loads(report)
     dependencies = report.get("vulnerable", {})  # type: ignore
     for dependency_name, dependency in dependencies.items():  # type: ignore
         for v in dependency["vulnerabilities"]:  # type: ignore
             references = [v["reference"]] + v["externalReferences"]
-            yield Issue(
+            yield VulnerabilityIssue(
                 cve=v["cve"],
                 cwe=v["cwe"],
                 description=v["description"],
@@ -145,10 +123,10 @@ def identify_pypi_references(
     )
 
 
-def from_pip_audit(report: str) -> Iterable[Issue]:
+def from_pip_audit(report: str) -> Iterable[VulnerabilityIssue]:
     """
     Transforms the JSON output from `nox -s dependency:audit` into an iterable of
-    `security.Issue` objects.
+    `VulnerabilityIssue` objects.
 
     This does not gracefully handle scenarios where:
      - a CVE is not initially associated with the vulnerability; however, the assumption
@@ -175,7 +153,7 @@ def from_pip_audit(report: str) -> Iterable[Issue]:
                 references=refs, package_name=package
             )
             if cves:
-                yield Issue(
+                yield VulnerabilityIssue(
                     cve=sorted(cves)[0],
                     cwe="None" if not cwes else ", ".join(cwes),
                     description=v["description"],
@@ -241,11 +219,11 @@ def _row(issue):
     return template.format(header=_header(), rows="\n".join(_row(i) for i in issues))
 
 
-def security_issue_title(issue: Issue) -> str:
+def security_issue_title(issue: VulnerabilityIssue) -> str:
     return f"🔐 {issue.cve}: {issue.coordinates}"
 
 
-def security_issue_body(issue: Issue) -> str:
+def security_issue_body(issue: VulnerabilityIssue) -> str:
     def as_markdown_listing(elements: Iterable[str]):
         return "\n".join(f"- {element}" for element in elements)
 
@@ -269,7 +247,9 @@ def as_markdown_listing(elements: Iterable[str]):
     )
 
 
-def create_security_issue(issue: Issue, project: str | None = None) -> tuple[str, str]:
+def create_security_issue(
+    issue: VulnerabilityIssue, project: str | None = None
+) -> tuple[str, str]:
     # fmt: off
     command = [
         "gh", "issue", "create",
@@ -322,14 +302,14 @@ def convert(
 
     def _maven(infile):
         issues = from_maven(infile.read())
-        for issue in _issues_as_json_str(issues):
-            stdout(issue)
+        for issue in issues:
+            stdout(issue.json_str)
         raise typer.Exit(code=0)
 
     def _pip_audit(infile):
         issues = from_pip_audit(infile.read())
-        for issue in _issues_as_json_str(issues):
-            stdout(issue)
+        for issue in issues:
+            stdout(issue.json_str)
         raise typer.Exit(code=0)
 
     actions = {Format.Maven: _maven, Format.PipAudit: _pip_audit}
@@ -364,10 +344,12 @@ def _github(infile):
         for issue in to_be_filtered:
             stderr(f"{issue}")
         filtered_issues = [
-            issue for issue in _issues(infile) if issue.cve not in to_be_filtered
+            issue
+            for issue in VulnerabilityIssue.extract_from_jsonl(infile)
+            if issue.cve not in to_be_filtered
         ]
-        for issue in _issues_as_json_str(filtered_issues):
-            stdout(issue)
+        for issue in filtered_issues:
+            stdout(issue.json_str)
         raise typer.Exit(code=0)
 
     def _pass_through(infile):
@@ -398,10 +380,13 @@ def create(
     Output:
     { "cve": "<cve-id>", "cwe": "<cwe-id>", "description": "<multiline string>", "coordinates": "<string>", "references": ["<url>", "<url>", ...], "issue_url": "<url>" }
     """
-    for issue in _issues(input_file):
+    for issue in VulnerabilityIssue.extract_from_jsonl(input_file):
         std_err, issue_url = create_security_issue(issue, project)
         stderr(std_err)
-        stdout(format_jsonl(issue_url, issue))
+        ghvi = GitHubVulnerabilityIssue.from_vulnerability_issue(
+            issue=issue, issue_url=issue_url
+        )
+        stdout(ghvi.json_str)
 
 
 class PPrintFormats(str, Enum):
@@ -421,11 +406,5 @@ def json_issue_to_markdown(
     print(issues_to_markdown(issues))
 
 
-def format_jsonl(issue_url: str, issue: Issue) -> str:
-    issue_json = asdict(issue)
-    issue_json["issue_url"] = issue_url.strip()
-    return json.dumps(issue_json)
-
-
 if __name__ == "__main__":
     CLI()

test/unit/conftest.py

@@ -7,7 +7,7 @@
 
 @pytest.fixture(scope="session")
 def pip_audit_jinja2_issue():
-    return security.Issue(
+    return security.VulnerabilityIssue(
         cve="CVE-2025-27516",
         cwe="None",
         description=cleandoc(
@@ -33,7 +33,7 @@ def pip_audit_jinja2_issue():
 
 @pytest.fixture(scope="session")
 def pip_audit_cryptography_issue():
-    return security.Issue(
+    return security.VulnerabilityIssue(
         cve="CVE-2024-12797",
         cwe="None",
         description=cleandoc(