Various changes

* Add basic filter
* Change CLI entry points
* Add prototype for issue creation

exasol/toolbox/tools/security.py

@@ -1,7 +1,12 @@
 import json
+import re
+import subprocess
 import sys
+from inspect import cleandoc
 from typing import (
+    Generator,
     Iterable,
+    Tuple,
 )
 
 import typer
@@ -10,24 +15,88 @@
 stdout = Console()
 stderr = Console(stderr=True)
 
-CLI = typer.Typer()
+ISSUE_CLI = typer.Typer()
+
+from dataclasses import (
+    asdict,
+    dataclass,
+)
 
-from dataclasses import dataclass, asdict
 
+def gh_security_issues() -> Generator[Tuple[str, str], None, None]:
+    """
+    Yields issue-id, cve-id pairs for all (closed, open) issues associated with CVEs
 
+    Return:
+        A generator which yield tuples of (id,title).
+
+    Raises:
+        subprocess.CalledProcessError: If the underlying command fails.
+    """
+    command = [
+        "gh",
+        "issue",
+        "list",
+        "--label",
+        "security",
+        "--search",
+        "CVE",
+        "--json",
+        "id,title",
+        "--limit",
+        "1000",
+        "--state",
+        "all",
+    ]
+    try:
+        result = subprocess.run(command, check=True, capture_output=True)
+    except FileNotFoundError as ex:
+        msg = "Command 'gh' not found. Please make sure you have installed the github cli."
+        raise FileNotFoundError(msg) from ex
+    except subprocess.CalledProcessError as ex:
+        stderr.print(f"{ex}")
+        raise ex
+
+    cve_pattern = re.compile(r"CVE-\d{4}-\d{4,7}")
+    issues = json.loads(result.stdout.decode("utf-8"))
+    issues = (
+        (issue["id"], cve_pattern.search(issue["title"]).group())  # type: ignore
+        for issue in issues
+        if cve_pattern.search(issue["title"])
+    )
+    return issues
+
+
+# Note:
+# In the long term we may want to adapt the official CVE json schema,
+# support for this could be generated using pydantic.
+# See here: https://github.com/CVEProject/cve-schema/blob/master/schema/v5.0/CVE_JSON_5.0_schema.json
 @dataclass(frozen=True)
 class Issue:
+    # Note: Add support additional (custom) information e.g. dependency tree etc.
     cve: str
     cwe: str
     description: str
     coordinates: str
     references: tuple
-    # Note: Add support additional (custom) information e.g. dependency tree etc.
+
+
+def _issues(input) -> Generator[Issue, None, None]:
+    issues = input.read()
+    issues = (line for line in issues.split("\n"))
+    issues = (json.loads(raw) for raw in issues)
+    issues = (Issue(**obj) for obj in issues)
+    yield from issues
+
+
+def _issues_as_json_str(issues):
+    for issue in issues:
+        issue = asdict(issue)  # type: ignore
+        yield json.dumps(issue)
 
 
 def from_maven(report: str) -> Iterable[Issue]:
-    # Notes:
-    # * Consider adding warnings if there is the same cve with multiple cooardinates
+    # Note: Consider adding warnings if there is the same cve with multiple coordinates
     report = json.loads(report)
     dependencies = report["vulnerable"]  # type: ignore
     for _, dependency in dependencies.items():  # type: ignore
@@ -42,51 +111,82 @@ def from_maven(report: str) -> Iterable[Issue]:
             )
 
 
-@CLI.command(name="convert")
+@ISSUE_CLI.command(name="convert")
 def convert(
         format: str = typer.Argument(..., help="input format to be converted."),
 ) -> None:
-    if format == 'maven':
+    if format == "maven":
         issues = from_maven(sys.stdin.read())
-        for issue in issues:
-            issue = asdict(issue)  # type: ignore
-            stdout.print(json.dumps(issue))
+        for issue in _issues_as_json_str(issues):
+            stdout.print(issue)
     else:
-        stderr.print("Unsupported format")
+        stderr.print(f"Unsupported format: {format}")
         sys.exit(-1)
 
 
-@CLI.command(name="filter")
+@ISSUE_CLI.command(name="filter")
 def filter(
-        _: str = typer.Argument(..., help="filter type to apply"),
+        type: str = typer.Argument(..., help="filter type to apply"),
 ) -> None:
-    for line in sys.stdin:
-        stdout.print(line, end='')
+    if type != "github":
+        stderr.print(
+            f"warning: Invalid filter type: {type}, falling back to pass through mode."
+        )
+        for line in sys.stdin:
+            stdout.print(line, end="")
+
+    to_be_filtered = list(gh_security_issues())
+    filtered_issues = [
+        issue for issue in _issues(sys.stdin) if issue.cve not in to_be_filtered
+    ]
+
+    for issue in _issues_as_json_str(filtered_issues):
+        stdout.print(issue)
 
 
-@CLI.command(name="create")
+@ISSUE_CLI.command(name="create")
 def create() -> None:
     for line in sys.stdin:
-        stdout.print(line, end='')
+        stdout.print(line, end="")
 
-    title = 'test'
-    body = 'another test'
+    title = "🔐 {cve}: {coordinates}"
+
+    body = cleandoc(
+        """
+        ## Summary
+        {description}
+
+        CVE: {cve}
+        CWE: {cwe}
+
+        ## References
+        {references}
+        """
+    )
+    body = "another test"
     command = [
-        'gh',
-        'issue',
-        'create',
-        '--label',
-        'security',
-        '--title',
-        title,
-        '--body',
-        body
+        "gh",
+        "issue",
+        "create",
+        "--label",
+        "security",
+        "--title",
+        title.format(cve="CVE-TEST", coordinates="some-package"),
+        "--body",
+        body.format(
+            cve="CVE-TEST",
+            cwe="CWE-TEST",
+            description="Some detailed description",
+            references="\n".join(f"- {ref}" for ref in ''),
+        ),
     ]
 
-    import subprocess
     result = subprocess.run(command, check=True)
     print(result)
 
 
+CLI = typer.Typer()
+CLI.add_typer(ISSUE_CLI, name='issuet s')
+
 if __name__ == "__main__":
     CLI()

pyproject.toml

@@ -102,4 +102,4 @@ ignore_errors = true
 
 [tool.poetry.scripts]
 tbx = 'exasol.toolbox.tools.tbx:CLI'
-security-issues = 'exasol.toolbox.tools.security:CLI'
+security = 'exasol.toolbox.tools.security:CLI'

test/unit/security_test.py

@@ -1,10 +1,66 @@
 import json
+import os
+import subprocess
+from contextlib import contextmanager
+from unittest import mock
 
 import pytest
 
 from exasol.toolbox.tools import security
 
 
+@contextmanager
+def empty_path():
+    """Make sure the PATH environment variable is empty."""
+    old_path = os.environ["PATH"]
+    os.environ["PATH"] = ""
+    yield
+    os.environ["PATH"] = old_path
+
+
+class TestGhSecurityIssues:
+    def test_gh_cli_is_not_available(self):
+        with empty_path():
+            with pytest.raises(FileNotFoundError) as exec_info:
+                set(security.gh_security_issues())
+
+        actual = f"{exec_info.value}"
+        expected = "Command 'gh' not found. Please make sure you have installed the github cli."
+
+        assert actual == expected
+
+    @mock.patch(
+        "subprocess.run",
+        side_effect=subprocess.CalledProcessError(
+            returncode=1, cmd=["command", "-l", "-v"]
+        ),
+    )
+    def test_gh_cli_failed(self, run_mock):
+        with pytest.raises(subprocess.CalledProcessError) as exec_info:
+            set(security.gh_security_issues())
+
+        actual = f"{exec_info.value}"
+        expected = f"{subprocess.CalledProcessError(returncode=1, cmd=['command', '-l', '-v'])}"
+        assert actual == expected
+
+    @mock.patch("subprocess.run")
+    def test_query_gh_security_issues(self, run_mock):
+        result = mock.MagicMock(subprocess.CompletedProcess)
+        result.returncode = 0
+        result.stdout = (
+            b'[{"id":"I_kwDOIRnUks5zutba","title":"\xf0\x9f\x90\x9e CVE-2023-41105: Fix build scripts "},'
+            b'{"id":"I_kwDOIRnUks5clFdR","title":"\xf0\x9f\x90\x9e CVE-2023-40217: Version check issues"}]\n'
+        )
+        run_mock.return_value = result
+
+        expected = {
+            ("I_kwDOIRnUks5zutba", "CVE-2023-41105"),
+            ("I_kwDOIRnUks5clFdR", "CVE-2023-40217"),
+        }
+        actual = set(security.gh_security_issues())
+        assert actual == expected
+
+
 @pytest.fixture()
 def maven_report():
     yield json.dumps(
@@ -98,13 +154,13 @@ def test_convert_maven_input(maven_report):
             cve="CVE-2023-39410",
             cwe="CWE-502",
             description="When deserializing untrusted or corrupted data, it is "
-            "possible for a reader to consume memory beyond the allowed "
-            "constraints and thus lead to out of memory on the system.\n"
-            "\n"
-            "This issue affects Java applications using Apache Avro "
-            "Java SDK up to and including 1.11.2.  Users should update "
-            "to apache-avro version 1.11.3 which addresses this issue.\n"
-            "\n",
+                        "possible for a reader to consume memory beyond the allowed "
+                        "constraints and thus lead to out of memory on the system.\n"
+                        "\n"
+                        "This issue affects Java applications using Apache Avro "
+                        "Java SDK up to and including 1.11.2.  Users should update "
+                        "to apache-avro version 1.11.3 which addresses this issue.\n"
+                        "\n",
             coordinates="pkg:maven/org.apache.avro/[email protected]",
             references=(
                 "https://ossindex.sonatype.org/vulnerability/CVE-2023-39410?component-type=maven&component-name=org.apache.avro%2Favro&utm_source=ossindex-client&utm_medium=integration&utm_content=1.8.1",
@@ -118,20 +174,20 @@ def test_convert_maven_input(maven_report):
             cve="CVE-2020-36641",
             cwe="CWE-611",
             description="A vulnerability classified as problematic was found in "
-            "gturri aXMLRPC up to 1.12.0. This vulnerability affects "
-            "the function ResponseParser of the file "
-            "src/main/java/de/timroes/axmlrpc/ResponseParser.java. The "
-            "manipulation leads to xml external entity reference. "
-            "Upgrading to version 1.12.1 is able to address this issue. "
-            "The patch is identified as "
-            "ad6615b3ec41353e614f6ea5fdd5b046442a832b. It is "
-            "recommended to upgrade the affected component. VDB-217450 "
-            "is the identifier assigned to this vulnerability.\n"
-            "\n"
-            "Sonatype's research suggests that this CVE's details "
-            "differ from those defined at NVD. See "
-            "https://ossindex.sonatype.org/vulnerability/CVE-2020-36641 "
-            "for details",
+                        "gturri aXMLRPC up to 1.12.0. This vulnerability affects "
+                        "the function ResponseParser of the file "
+                        "src/main/java/de/timroes/axmlrpc/ResponseParser.java. The "
+                        "manipulation leads to xml external entity reference. "
+                        "Upgrading to version 1.12.1 is able to address this issue. "
+                        "The patch is identified as "
+                        "ad6615b3ec41353e614f6ea5fdd5b046442a832b. It is "
+                        "recommended to upgrade the affected component. VDB-217450 "
+                        "is the identifier assigned to this vulnerability.\n"
+                        "\n"
+                        "Sonatype's research suggests that this CVE's details "
+                        "differ from those defined at NVD. See "
+                        "https://ossindex.sonatype.org/vulnerability/CVE-2020-36641 "
+                        "for details",
             coordinates="pkg:maven/fr.turri/[email protected]",
             references=(
                 "https://ossindex.sonatype.org/vulnerability/CVE-2020-36641?component-type=maven&component-name=fr.turri%2FaXMLRPC&utm_source=ossindex-client&utm_medium=integration&utm_content=1.8.1",