Allow for filtering dependencies and security warnings by the availab…

…ility of a fix or upgrade.

Closes #10323.

components/collector/src/base_collectors/source_collector.py

@@ -11,6 +11,8 @@
 import aiohttp
 from packaging.version import Version
 
+from shared_data_model import DATA_MODEL
+
 from collector_utilities.date_time import days_ago, days_to_go
 from collector_utilities.exceptions import CollectorError
 from collector_utilities.functions import (
@@ -300,7 +302,13 @@ def _include_entity(self, entity: Entity) -> bool:
         severity = str(entity[self.ENTITY_SEVERITY_ATTRIBUTE])
         if self.MAKE_ENTITY_SEVERITY_VALUE_LOWER_CASE:
             severity = severity.lower()
-        return severity in self._parameter(self.SEVERITY_PARAMETER) and super()._include_entity(entity)
+        if severity not in self._parameter(self.SEVERITY_PARAMETER):
+            return False
+        if "fix_status" in DATA_MODEL.sources[self.source_type].parameters:
+            fix_status = self._parameter("fix_status")
+            has_fix = entity["fix"] and entity["fix"] != "None"
+            return ("fix available" in fix_status and has_fix) or ("fix not available" in fix_status and not has_fix)
+        return super()._include_entity(entity)
 
 
 class TimeCollector(SourceCollector):

components/collector/src/source_collectors/anchore/security_warnings.py

@@ -2,27 +2,18 @@
 
 from shared.utils.functions import md5_hash
 
-from base_collectors import JSONFileSourceCollector
+from base_collectors import JSONFileSourceCollector, SecurityWarningsSourceCollector
 from collector_utilities.type import JSON
 from model import Entities, Entity
 
 
-class AnchoreSecurityWarnings(JSONFileSourceCollector):
+class AnchoreSecurityWarnings(JSONFileSourceCollector, SecurityWarningsSourceCollector):
     """Anchore collector for security warnings."""
 
     def _parse_json(self, json: JSON, filename: str) -> Entities:
         """Override to parse the Anchore security warnings."""
-        severities = self._parameter("severities")
-        entities = Entities()
         vulnerabilities = json.get("vulnerabilities", []) if isinstance(json, dict) else []
-        entities.extend(
-            [
-                self._create_entity(vulnerability, filename)
-                for vulnerability in vulnerabilities
-                if vulnerability["severity"] in severities
-            ],
-        )
-        return entities
+        return Entities([self._create_entity(vulnerability, filename) for vulnerability in vulnerabilities])
 
     @staticmethod
     def _create_entity(vulnerability: dict[str, str], filename: str) -> Entity:

components/collector/tests/source_collectors/anchore/base.py

@@ -13,11 +13,10 @@ def setUp(self):
         super().setUp()
         self.url = "https://cve"
         self.set_source_parameter("details_url", "image-details.json")
-        self.set_source_parameter("severities", ["Low"])
         self.vulnerabilities_json = {
             "vulnerabilities": [
                 {"vuln": "CVE-000", "package": "package", "fix": "None", "url": self.url, "severity": "Low"},
-                {"vuln": "CVE-000", "package": "package2", "fix": "None", "url": self.url, "severity": "Unknown"},
+                {"vuln": "CVE-000", "package": "package2", "fix": "v1.2.3", "url": self.url, "severity": "Unknown"},
             ],
         }
         self.details_json = [{"analyzed_at": "2020-02-07T22:53:43Z"}]

components/collector/tests/source_collectors/anchore/test_security_warnings.py

@@ -12,40 +12,43 @@ class AnchoreSecurityWarningsTest(AnchoreTestCase):
 
     METRIC_TYPE = "security_warnings"
 
+    def create_entity(
+        self, cve: str, package: str, fix: str = "None", severity: str = "Low", filename: str = ""
+    ) -> dict[str, str]:
+        """Return an expected entitiy."""
+        return {
+            "key": md5_hash(f"{filename}{cve}:{package}"),
+            "filename": filename,
+            "cve": cve,
+            "url": self.url,
+            "fix": fix,
+            "severity": severity,
+            "package": package,
+        }
+
     async def test_warnings(self):
         """Test the number of security warnings."""
+        self.set_source_parameter("severities", ["Low"])
+        response = await self.collect(get_request_json_return_value=self.vulnerabilities_json)
+        expected_entities = [self.create_entity("CVE-000", "package")]
+        self.assert_measurement(response, value="1", entities=expected_entities)
+
+    async def test_filter_by_fix_status(self):
+        """Test the security warnings can be filtered by fix status."""
+        self.set_source_parameter("fix_status", ["fix available"])
         response = await self.collect(get_request_json_return_value=self.vulnerabilities_json)
-        expected_entities = [
-            {
-                "key": md5_hash("CVE-000:package"),
-                "filename": "",
-                "cve": "CVE-000",
-                "url": self.url,
-                "fix": "None",
-                "severity": "Low",
-                "package": "package",
-            },
-        ]
+        expected_entities = [self.create_entity("CVE-000", package="package2", fix="v1.2.3", severity="Unknown")]
         self.assert_measurement(response, value="1", entities=expected_entities)
 
     async def test_zipped_report(self):
         """Test that a zip with reports can be read."""
+        self.set_source_parameter("severities", ["Low"])
         self.set_source_parameter("url", "anchore.zip")
         filename = "vuln.json"
         zipfile = self.zipped_report(
             (filename, json.dumps(self.vulnerabilities_json)),
             ("details.json", json.dumps(self.details_json)),
         )
         response = await self.collect(get_request_content=zipfile)
-        expected_entities = [
-            {
-                "key": md5_hash(f"{filename}CVE-000:package"),
-                "filename": filename,
-                "cve": "CVE-000",
-                "url": self.url,
-                "fix": "None",
-                "severity": "Low",
-                "package": "package",
-            },
-        ]
+        expected_entities = [self.create_entity("CVE-000", "package", filename=filename)]
         self.assert_measurement(response, value="1", entities=expected_entities)

components/collector/tests/source_collectors/anchore_jenkins_plugin/test_security_warnings.py

@@ -11,27 +11,62 @@ class AnchoreJenkinsPluginSecurityWarningsTest(SourceCollectorTestCase):
     METRIC_TYPE = "security_warnings"
     SOURCE_TYPE = "anchore_jenkins_plugin"
 
-    async def test_warnings(self):
-        """Test the number of security warnings."""
-        self.set_source_parameter("severities", ["Low"])
-        vulnerabilities_json = {
+    def setUp(self) -> None:
+        """Create the Anchore Jenkins plugin JSON fixture."""
+        super().setUp()
+        self.vulnerabilities_json = {
             "data": [
                 ["tag", "CVE-000", "Low", "package", "None", "https://cve-000"],
-                ["tag", "CVE-001", "Unknown", "package2", "None", "https://cve-001"],
+                ["tag", "CVE-001", "Unknown", "package2", "v1.2.3", "https://cve-001"],
             ],
         }
+
+    def create_entity(self, cve: str, package: str, severity: str = "Low", fix: str = "None") -> dict[str, str]:
+        """Create an expected entity."""
+        return {
+            "key": md5_hash(f"tag:{cve}:{package}"),
+            "tag": "tag",
+            "cve": cve,
+            "url": f"https://{cve.lower()}",
+            "fix": fix,
+            "severity": severity,
+            "package": package,
+        }
+
+    async def test_warnings(self):
+        """Test the number of security warnings."""
         response = await self.collect(
-            get_request_json_side_effect=[{"name": "job", "lastSuccessfulBuild": {"number": 42}}, vulnerabilities_json],
+            get_request_json_side_effect=[
+                {"name": "job", "lastSuccessfulBuild": {"number": 42}},
+                self.vulnerabilities_json,
+            ],
         )
         expected_entities = [
-            {
-                "key": md5_hash("tag:CVE-000:package"),
-                "tag": "tag",
-                "cve": "CVE-000",
-                "url": "https://cve-000",
-                "fix": "None",
-                "severity": "Low",
-                "package": "package",
-            },
+            self.create_entity("CVE-000", "package"),
+            self.create_entity("CVE-001", "package2", severity="Unknown", fix="v1.2.3"),
         ]
+        self.assert_measurement(response, value="2", entities=expected_entities)
+
+    async def test_filter_warnings_by_severity(self):
+        """Test the number of security warnings."""
+        self.set_source_parameter("severities", ["Low"])
+        response = await self.collect(
+            get_request_json_side_effect=[
+                {"name": "job", "lastSuccessfulBuild": {"number": 42}},
+                self.vulnerabilities_json,
+            ],
+        )
+        expected_entities = [self.create_entity("CVE-000", "package")]
+        self.assert_measurement(response, value="1", entities=expected_entities)
+
+    async def test_filter_warnings_by_dix_status(self):
+        """Test the number of security warnings."""
+        self.set_source_parameter("fix_status", ["fix not available"])
+        response = await self.collect(
+            get_request_json_side_effect=[
+                {"name": "job", "lastSuccessfulBuild": {"number": 42}},
+                self.vulnerabilities_json,
+            ],
+        )
+        expected_entities = [self.create_entity("CVE-000", "package")]
         self.assert_measurement(response, value="1", entities=expected_entities)

components/shared_code/src/shared_data_model/parameters.py

@@ -105,6 +105,16 @@ class Days(IntegerParameter):
     min_value: str = "1"
 
 
+class FixStatus(MultipleChoiceParameter):
+    """Fix status for security warnings."""
+
+    name: str = "Fix status"
+    placeholder: str | None = "all fix statuses"
+    metrics: list[str] = ["security_warnings"]
+    help: str | None = "Show security warnings without fix, with fix, or both."
+    values: list[str] | None = ["fix available", "fix not available"]
+
+
 class Severities(MultipleChoiceParameter):
     """Security warning severities."""
 

components/shared_code/src/shared_data_model/sources/anchore.py

@@ -4,7 +4,7 @@
 
 from shared_data_model.meta.entity import Color, Entity, EntityAttribute
 from shared_data_model.meta.source import Source
-from shared_data_model.parameters import URL, Severities, access_parameters
+from shared_data_model.parameters import URL, FixStatus, Severities, access_parameters
 
 from .jenkins import JENKINS_TOKEN_DOCS, jenkins_access_parameters
 
@@ -33,6 +33,7 @@
             metrics=["source_up_to_dateness"],
         ),
         "severities": SEVERITIES,
+        "fix_status": FixStatus(),
         **access_parameters(
             ALL_ANCHORE_METRICS,
             source_type="an Anchore vulnerability report",
@@ -55,6 +56,7 @@
     url=HttpUrl("https://plugins.jenkins.io/anchore-container-scanner/"),
     parameters={
         "severities": SEVERITIES,
+        "fix_status": FixStatus(),
         **jenkins_access_parameters(
             ALL_ANCHORE_METRICS,
             kwargs={

docs/src/changelog.md

@@ -18,12 +18,18 @@ If your currently installed *Quality-time* version is not the latest version, pl
 
 - When measuring test cases with Visual Studio TRX as source, search all test category items for test case ids, instead of cutting the search short after the first match. Fixes [#10460](https://github.com/ICTU/quality-time/issues/10460).
 
+### Added
+
+- Allow for filtering dependencies and security warnings by the availability of a fix or upgrade. Closes [#10323](https://github.com/ICTU/quality-time/issues/10323).
+
 ## v5.20.0 - 2024-12-05
 
 ### Added
 
 - Add a new metric 'test suites' that can be used to count test suites (or test scenarios in Robot Framework parlance). Test suites can be filtered by test result. Supporting sources are Robot Framework, JUnit, and TestNG. Closes [#10078](https://github.com/ICTU/quality-time/issues/10078).
 
+## v5.20.0 - 2024-12-05
+
 ### Changed
 
 - Change the 'unmerged branches' metric to 'inactive branches', also enabling it to count branches that have been merged but not deleted. Closes [#1253](https://github.com/ICTU/quality-time/issues/1253).