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 that the security warnings can be filtered by severity."""
+        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_fix_status(self):
+        """Test that the security warnings can be filtered by fix status."""
+        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,6 +18,10 @@ 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