Time since last SBoM analysis in Dependency-Track.

Allow for measuring the time since the last analysis date of an SBoM in Dependency-Track using the 'source up-to-dateness' metric. This can be configured using the new 'event type' parameter. Possible values are 'last SBOM upload', 'last SBOM analysis', or 'both (oldest of last SBOM analysis and last SBOM upload)'.

Closes #9764.

components/collector/.vulture_ignore_list.py

@@ -37,14 +37,16 @@
 CxSASTSecurityWarnings  # unused class (src/source_collectors/cxsast/security_warnings.py:10)
 CxSASTSourceUpToDateness  # unused class (src/source_collectors/cxsast/source_up_to_dateness.py:10)
 CxSASTSourceVersion  # unused class (src/source_collectors/cxsast/source_version.py:12)
-lastBomImport  # unused variable (src/source_collectors/dependency_track/base.py:17)
+lastOccurrence  # unused variable (src/source_collectors/dependency_track/base.py:16)
+lastBomImport  # unused variable (src/source_collectors/dependency_track/base.py:24)
 latestVersion  # unused variable (src/source_collectors/dependency_track/dependencies.py:14)
 repositoryMeta  # unused variable (src/source_collectors/dependency_track/dependencies.py:22)
 DependencyTrackDependencies  # unused class (src/source_collectors/dependency_track/dependencies.py:27)
-vulnId  # unused variable (src/source_collectors/dependency_track/security_warnings.py:23)
-matrix  # unused variable (src/source_collectors/dependency_track/security_warnings.py:34)
-DependencyTrackSecurityWarnings  # unused class (src/source_collectors/dependency_track/security_warnings.py:38)
-DependencyTrackSourceUpToDateness  # unused class (src/source_collectors/dependency_track/source_up_to_dateness.py:14)
+latestVersion  # unused variable (src/source_collectors/dependency_track/security_warnings.py:15)
+vulnId  # unused variable (src/source_collectors/dependency_track/security_warnings.py:25)
+matrix  # unused variable (src/source_collectors/dependency_track/security_warnings.py:36)
+DependencyTrackSecurityWarnings  # unused class (src/source_collectors/dependency_track/security_warnings.py:40)
+DependencyTrackSourceUpToDateness  # unused class (src/source_collectors/dependency_track/source_up_to_dateness.py:15)
 DependencyTrackSourceVersion  # unused class (src/source_collectors/dependency_track/source_version.py:9)
 GatlingPerformanceTestDuration  # unused class (src/source_collectors/gatling/performancetest_duration.py:9)
 GatlingSlowTransactions  # unused class (src/source_collectors/gatling/slow_transactions.py:12)
@@ -70,7 +72,7 @@
 JacocoJenkinsPluginSourceUpToDateness  # unused class (src/source_collectors/jacoco_jenkins_plugin/source_up_to_dateness.py:8)
 JacocoJenkinsPluginUncoveredBranches  # unused class (src/source_collectors/jacoco_jenkins_plugin/uncovered_branches.py:6)
 JacocoJenkinsPluginUncoveredLines  # unused class (src/source_collectors/jacoco_jenkins_plugin/uncovered_lines.py:6)
-buildable  # unused variable (src/source_collectors/jenkins/base.py:29)
+buildable  # unused variable (src/source_collectors/jenkins/base.py:30)
 JenkinsChangeFailureRate  # unused class (src/source_collectors/jenkins/change_failure_rate.py:12)
 JenkinsFailedJobs  # unused class (src/source_collectors/jenkins/failed_jobs.py:8)
 JenkinsJobRunsWithinTimePeriod  # unused class (src/source_collectors/jenkins/job_runs_within_time_period.py:11)
@@ -152,9 +154,9 @@
 References  # unused variable (src/source_collectors/trivy/security_warnings.py:26)
 Target  # unused variable (src/source_collectors/trivy/security_warnings.py:32)
 Vulnerabilities  # unused variable (src/source_collectors/trivy/security_warnings.py:33)
-SchemaVersion  # unused variable (src/source_collectors/trivy/security_warnings.py:42)
-Results  # unused variable (src/source_collectors/trivy/security_warnings.py:43)
-TrivyJSONSecurityWarnings  # unused class (src/source_collectors/trivy/security_warnings.py:49)
+SchemaVersion  # unused variable (src/source_collectors/trivy/security_warnings.py:49)
+Results  # unused variable (src/source_collectors/trivy/security_warnings.py:50)
+TrivyJSONSecurityWarnings  # unused class (src/source_collectors/trivy/security_warnings.py:56)
 totalCount  # unused variable (tests/source_collectors/github/test_merge_requests.py:16)
 baseRefName  # unused variable (tests/source_collectors/github/test_merge_requests.py:24)
 createdAt  # unused variable (tests/source_collectors/github/test_merge_requests.py:27)

components/collector/src/source_collectors/dependency_track/base.py

@@ -1,14 +1,21 @@
 """Dependency-Track base collector."""
 
 from collections.abc import AsyncIterator
-from typing import Literal, TypedDict, cast
+from typing import Literal, NotRequired, TypedDict, cast
 
 from base_collectors import SourceCollector
 from collector_utilities.functions import add_query, match_string_or_regular_expression
 from collector_utilities.type import URL, Response
 from model import Entity, SourceResponses
 
 
+class DependencyTrackMetrics(TypedDict, total=False):
+    """Project metrics as returned by Dependency-Track."""
+
+    # Last occurrence is a Unix timestamp of the datetime of the last BOM analysis
+    lastOccurrence: int
+
+
 class DependencyTrackProject(TypedDict):
     """Project as returned by Dependency-Track."""
 
@@ -18,6 +25,7 @@ class DependencyTrackProject(TypedDict):
     name: str
     uuid: str
     version: str
+    metrics: NotRequired[DependencyTrackMetrics]
 
 
 class DependencyTrackBase(SourceCollector):

components/collector/src/source_collectors/dependency_track/source_up_to_dateness.py

@@ -21,11 +21,11 @@ async def _api_url(self) -> URL:
 
     async def _parse_source_response_date_times(self, responses: SourceResponses) -> Sequence[datetime]:
         """Override to parse the timestamp from the responses."""
-        if datetimes := [
-            self._last_bom_import_datetime(project)
-            for response in responses
-            async for project in self._get_projects_from_response(response)
-        ]:
+        datetimes = []
+        for response in responses:
+            async for project in self._get_projects_from_response(response):
+                datetimes.extend(self._project_event_datetimes(project))
+        if datetimes:
             return datetimes
         error_message = "No projects found"
         raise CollectorError(error_message)
@@ -36,17 +36,40 @@ async def _parse_entities(self, responses: SourceResponses) -> Entities:
         entities = Entities()
         for response in responses:
             async for project in self._get_projects_from_response(response):
+                last_bom_analysis = self._last_bom_analysis_datetime(project)
+                last_bom_import = self._last_bom_import_datetime(project)
                 uuid = project["uuid"]
                 entity = Entity(
                     key=uuid,
-                    last_bom_import=self._last_bom_import_datetime(project).isoformat(),
+                    last_bom_analysis="" if last_bom_analysis is None else last_bom_analysis.isoformat(),
+                    last_bom_import="" if last_bom_import is None else last_bom_import.isoformat(),
                     project=project["name"],
                     project_landing_url=f"{landing_url}/projects/{uuid}",
                 )
                 entities.append(entity)
         return entities
 
-    @staticmethod
-    def _last_bom_import_datetime(project: DependencyTrackProject) -> datetime:
+    def _project_event_datetimes(self, project: DependencyTrackProject) -> list[datetime]:
+        """Return the project event (BOM import, BOM analysis) datetimes."""
+        datetimes = []
+        event_types = self._parameter("project_event_types")
+        if "last BOM import" in event_types and (datetime := self._last_bom_import_datetime(project)):
+            datetimes.append(datetime)
+        if "last BOM analysis" in event_types and (datetime := self._last_bom_analysis_datetime(project)):
+            datetimes.append(datetime)
+        return datetimes
+
+    @classmethod
+    def _last_bom_import_datetime(cls, project: DependencyTrackProject) -> datetime | None:
         """Return the last BOM import datetime for the project."""
-        return datetime_from_timestamp(int(project["lastBomImport"]))
+        return cls._timestamp_to_datetime(project.get("lastBomImport"))
+
+    @classmethod
+    def _last_bom_analysis_datetime(cls, project: DependencyTrackProject) -> datetime | None:
+        """Return the last BOM analysis datetime for the project."""
+        return cls._timestamp_to_datetime(project.get("metrics", {}).get("lastOccurrence"))
+
+    @staticmethod
+    def _timestamp_to_datetime(timestamp: int | None) -> datetime | None:
+        """Convert the timestamp to a datetime."""
+        return None if timestamp is None else datetime_from_timestamp(timestamp)

components/collector/tests/source_collectors/dependency_track/base_test.py

@@ -0,0 +1,23 @@
+"""Base classes for Dependency-Track collector unit tests."""
+
+from source_collectors.dependency_track.base import DependencyTrackMetrics, DependencyTrackProject
+
+from tests.source_collectors.source_collector_test_case import SourceCollectorTestCase
+
+
+class DependencyTrackTestCase(SourceCollectorTestCase):
+    """Base class for Dependency-Track collector Unit tests."""
+
+    SOURCE_TYPE = "dependency_track"
+
+    def projects(self) -> list[DependencyTrackProject]:
+        """Create the Dependency-Track projects fixture."""
+        return [
+            DependencyTrackProject(
+                name="project name",
+                uuid="project uuid",
+                version="1.4",
+                lastBomImport=0,
+                metrics=DependencyTrackMetrics(),
+            ),
+        ]

components/collector/tests/source_collectors/dependency_track/test_dependencies.py

@@ -1,20 +1,14 @@
 """Unit tests for the Dependency-Track security warnings collector."""
 
-from source_collectors.dependency_track.base import DependencyTrackProject
 from source_collectors.dependency_track.dependencies import DependencyTrackComponent
 
-from tests.source_collectors.source_collector_test_case import SourceCollectorTestCase
+from .base_test import DependencyTrackTestCase
 
 
-class DependencyTrackDependenciesTest(SourceCollectorTestCase):
+class DependencyTrackDependenciesTest(DependencyTrackTestCase):
     """Unit tests for the Dependency-Track dependencies collector."""
 
     METRIC_TYPE = "dependencies"
-    SOURCE_TYPE = "dependency_track"
-
-    def projects(self) -> list[DependencyTrackProject]:
-        """Create the Dependency-Track projects fixture."""
-        return [DependencyTrackProject(name="project name", uuid="project uuid", version="1.4", lastBomImport=0)]
 
     def dependencies(self, latest_version: str) -> list[DependencyTrackComponent]:
         """Create a list of dependencies as returned by Dependency-Track."""

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

@@ -2,25 +2,20 @@
 
 from aiohttp import BasicAuth
 
-from source_collectors.dependency_track.base import DependencyTrackBase, DependencyTrackProject
+from source_collectors.dependency_track.base import DependencyTrackBase
 from source_collectors.dependency_track.security_warnings import (
     DependencyTrackComponent,
     DependencyTrackFinding,
     DependencyTrackVulnerability,
 )
 
-from tests.source_collectors.source_collector_test_case import SourceCollectorTestCase
+from .base_test import DependencyTrackTestCase
 
 
-class DependencyTrackSecurityWarningsTest(SourceCollectorTestCase):
+class DependencyTrackSecurityWarningsTest(DependencyTrackTestCase):
     """Unit tests for the Dependency-Track security warnings collector."""
 
     METRIC_TYPE = "security_warnings"
-    SOURCE_TYPE = "dependency_track"
-
-    def projects(self) -> list[DependencyTrackProject]:
-        """Create the Dependency-Track projects fixture."""
-        return [DependencyTrackProject(name="project name", uuid="project uuid", version="1.4", lastBomImport=0)]
 
     def findings(self) -> list[DependencyTrackFinding]:
         """Create the Dependency-Track findings fixture."""

components/collector/tests/source_collectors/dependency_track/test_source_up_to_dateness.py

@@ -4,42 +4,63 @@
 
 from dateutil.tz import tzlocal
 
-from source_collectors.dependency_track.base import DependencyTrackBase, DependencyTrackProject
+from source_collectors.dependency_track.base import (
+    DependencyTrackBase,
+    DependencyTrackMetrics,
+    DependencyTrackProject,
+)
 
-from tests.source_collectors.source_collector_test_case import SourceCollectorTestCase
+from .base_test import DependencyTrackTestCase
 
 
-class DependencyTrackSourceUpToDatenessVersionTest(SourceCollectorTestCase):
+class DependencyTrackSourceUpToDatenessVersionTest(DependencyTrackTestCase):
     """Unit tests for the Dependency-Track source up-to-dateness collector."""
 
     METRIC_ADDITION = "min"
     METRIC_TYPE = "source_up_to_dateness"
-    SOURCE_TYPE = "dependency_track"
     LANDING_URL = "https://dependency_track"
 
     def projects(self) -> list[DependencyTrackProject]:
         """Create Dependency-Track projects fixture."""
         now = datetime.now(tz=tzlocal()).replace(microsecond=0)
         self.yesterday = now - timedelta(days=1)
         yesterday_timestamp = round(self.yesterday.timestamp() * 1000)
+        self.day_before_yesterday = now - timedelta(days=2)
+        day_before_yesterday_timestamp = round(self.day_before_yesterday.timestamp() * 1000)
         self.last_week = now - timedelta(days=7)
-        last_week_timestamp = round(self.last_week.timestamp() * 1000)
+        self.last_week_timestamp = round(self.last_week.timestamp() * 1000)
+        self.few_days_ago = now - timedelta(days=4)
+        self.few_days_ago_timestamp = round(self.few_days_ago.timestamp() * 1000)
         return [
-            DependencyTrackProject(name="Project 1", version="1.1", uuid="p1", lastBomImport=yesterday_timestamp),
-            DependencyTrackProject(name="Project 2", version="1.2", uuid="p2", lastBomImport=last_week_timestamp),
+            DependencyTrackProject(
+                name="Project 1",
+                version="1.1",
+                uuid="p1",
+                lastBomImport=yesterday_timestamp,
+                metrics=DependencyTrackMetrics(lastOccurrence=self.few_days_ago_timestamp),
+            ),
+            DependencyTrackProject(
+                name="Project 2",
+                version="1.2",
+                uuid="p2",
+                lastBomImport=self.last_week_timestamp,
+                metrics=DependencyTrackMetrics(lastOccurrence=day_before_yesterday_timestamp),
+            ),
         ]
 
     def entities(self) -> list[dict[str, str]]:
         """Create the expected entities."""
         return [
             {
                 "key": "p1",
+                "last_bom_analysis": self.few_days_ago.isoformat(),
                 "last_bom_import": self.yesterday.isoformat(),
                 "project": "Project 1",
                 "project_landing_url": "/projects/p1",
             },
             {
                 "key": "p2",
+                "last_bom_analysis": self.day_before_yesterday.isoformat(),
                 "last_bom_import": self.last_week.isoformat(),
                 "project": "Project 2",
                 "project_landing_url": "/projects/p2",
@@ -51,6 +72,54 @@ async def test_source_up_to_dateness(self):
         response = await self.collect(get_request_json_return_value=self.projects())
         self.assert_measurement(response, value="7", landing_url=self.LANDING_URL, entities=self.entities())
 
+    async def test_source_up_to_dateness_missing_bom_analysis(self):
+        """Test that the source up-to-dateness can be measured even if the BOM has no last analysis datetime."""
+        projects = self.projects()
+        projects.append(
+            DependencyTrackProject(
+                name="Project 3",
+                version="1.3",
+                uuid="p3",
+                lastBomImport=self.last_week_timestamp,
+                metrics=DependencyTrackMetrics(),
+            )
+        )
+        entities = self.entities()
+        entities.append(
+            {
+                "key": "p3",
+                "last_bom_analysis": "",
+                "last_bom_import": self.last_week.isoformat(),
+                "project": "Project 3",
+                "project_landing_url": "/projects/p3",
+            }
+        )
+        response = await self.collect(get_request_json_return_value=projects)
+        self.assert_measurement(response, value="7", landing_url=self.LANDING_URL, entities=entities)
+
+    async def test_source_up_to_dateness_missing_bom_import(self):
+        """Test that the source up-to-dateness can be measured even if the BOM has no import datetime."""
+        projects = self.projects()
+        projects.append(
+            DependencyTrackProject(
+                name="Project 3",
+                version="1.3",
+                uuid="p3",
+            )
+        )
+        entities = self.entities()
+        entities.append(
+            {
+                "key": "p3",
+                "last_bom_analysis": "",
+                "last_bom_import": "",
+                "project": "Project 3",
+                "project_landing_url": "/projects/p3",
+            }
+        )
+        response = await self.collect(get_request_json_return_value=projects)
+        self.assert_measurement(response, value="7", landing_url=self.LANDING_URL, entities=entities)
+
     async def test_filter_by_project_name(self):
         """Test that projects can be filtered by name."""
         self.set_source_parameter("project_names", ["other project"])
@@ -76,6 +145,18 @@ async def test_filter_by_project_name_and_version(self):
         response = await self.collect(get_request_json_return_value=self.projects())
         self.assert_measurement(response, parse_error="No projects found")
 
+    async def test_filter_by_event_type_last_bom_import(self):
+        """Test that projects can be filtered by event type."""
+        self.set_source_parameter("project_event_types", ["last BOM import"])
+        response = await self.collect(get_request_json_return_value=self.projects())
+        self.assert_measurement(response, value="7", landing_url=self.LANDING_URL, entities=self.entities())
+
+    async def test_filter_by_event_type_last_bom_analysis(self):
+        """Test that projects can be filtered by event type."""
+        self.set_source_parameter("project_event_types", ["last BOM analysis"])
+        response = await self.collect(get_request_json_return_value=self.projects())
+        self.assert_measurement(response, value="4", landing_url=self.LANDING_URL, entities=self.entities())
+
     async def test_source_up_to_dateness_with_pagination(self):
         """Test that the source up-to-dateness can be measured when pagination is needed."""
         default_size = DependencyTrackBase.PAGE_SIZE

components/collector/tests/source_collectors/dependency_track/test_source_version.py

@@ -1,14 +1,13 @@
 """Unit tests for the Dependency-Track source version collector."""
 
-from tests.source_collectors.source_collector_test_case import SourceCollectorTestCase
+from .base_test import DependencyTrackTestCase
 
 
-class DependencyTrackSourceVersionTest(SourceCollectorTestCase):
+class DependencyTrackSourceVersionTest(DependencyTrackTestCase):
     """Unit tests for the Dependency-Track source version collector."""
 
     METRIC_ADDITION = "min"
     METRIC_TYPE = "source_version"
-    SOURCE_TYPE = "dependency_track"
 
     async def test_source_version(self):
         """Test that the source version can be measured."""

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

@@ -78,6 +78,15 @@
             metrics=["dependencies", "security_warnings"],
         ),
         "severities": Severities(values=["Unassigned", "Info", "Low", "Medium", "High", "Critical"]),
+        "project_event_types": MultipleChoiceParameter(
+            name="Project event types",
+            short_name="event types",
+            placeholder="all event types",
+            default_value=["last BOM import"],
+            help="Project event types to consider for measuring the up-to-dateness.",
+            values=["last BOM analysis", "last BOM import"],
+            metrics=["source_up_to_dateness"],
+        ),
     },
     entities={
         "dependencies": Entity(
@@ -105,6 +114,7 @@
             attributes=[
                 EntityAttribute(name="Project", url="project_landing_url"),
                 EntityAttribute(name="Last BOM import", type=EntityAttributeType.DATETIME),
+                EntityAttribute(name="Last BOM analysis", type=EntityAttributeType.DATETIME),
             ],
         ),
     },