Merge pull request #310 from DependencyTrack/issue-782-add-token-in-n…

…otifications

Added correlation token in workflow notification subjects

src/main/java/org/dependencytrack/event/kafka/KafkaStreamsTopologyFactory.java

@@ -170,9 +170,12 @@ Topology createTopology() {
                     try {
                         alpineNotification = vulnScan.getStatus() == VulnerabilityScan.Status.FAILED
                                 ? createProjectVulnerabilityAnalysisCompleteNotification(vulnScan,
+                                UUID.fromString(scanToken),
                                 ProjectVulnAnalysisStatus.PROJECT_VULN_ANALYSIS_STATUS_FAILED)
                                 : createProjectVulnerabilityAnalysisCompleteNotification(
-                                vulnScan, ProjectVulnAnalysisStatus.PROJECT_VULN_ANALYSIS_STATUS_COMPLETED);
+                                vulnScan,
+                                UUID.fromString(scanToken),
+                                ProjectVulnAnalysisStatus.PROJECT_VULN_ANALYSIS_STATUS_COMPLETED);
                     } catch (RuntimeException e) {
                         LOGGER.warn("Failed to generate a %s notification (project: %s; token: %s)"
                                 .formatted(NotificationGroup.PROJECT_VULN_ANALYSIS_COMPLETE,

src/main/java/org/dependencytrack/event/kafka/processor/DelayedBomProcessedNotificationProcessor.java

@@ -65,7 +65,7 @@ public void process(final Record<String, VulnerabilityScan> record) {
                 // BOM format and spec version are hardcoded because we don't have this information at this point.
                 // DT currently only accepts CycloneDX anyway.
                 .content("A %s BOM was processed".formatted(Bom.Format.CYCLONEDX.getFormatShortName()))
-                .subject(new BomConsumedOrProcessed(project, /* bom */ "(Omitted)", Bom.Format.CYCLONEDX, "Unknown"));
+                .subject(new BomConsumedOrProcessed(UUID.fromString(vulnScan.getToken()), project, /* bom */ "(Omitted)", Bom.Format.CYCLONEDX, "Unknown"));
 
         context().forward(record.withKey(project.getUuid().toString()).withValue(convert(alpineNotification)));
         LOGGER.info("Dispatched delayed %s notification (token=%s, project=%s)"

src/main/java/org/dependencytrack/notification/vo/BomConsumedOrProcessed.java

@@ -21,20 +21,28 @@
 import org.dependencytrack.model.Bom;
 import org.dependencytrack.model.Project;
 
+import java.util.UUID;
+
 public class BomConsumedOrProcessed {
 
+    private UUID token;
     private Project project;
     private String bom;
     private Bom.Format format;
     private String specVersion;
 
-    public BomConsumedOrProcessed(final Project project, final String bom, final Bom.Format format, final String specVersion) {
+    public BomConsumedOrProcessed(final UUID token, final Project project, final String bom, final Bom.Format format, final String specVersion) {
+        this.token = token;
         this.project = project;
         this.bom = bom;
         this.format = format;
         this.specVersion = specVersion;
     }
 
+    public UUID getToken() {
+        return token;
+    }
+
     public Project getProject() {
         return project;
     }

src/main/java/org/dependencytrack/notification/vo/BomProcessingFailed.java

@@ -21,22 +21,30 @@
 import org.dependencytrack.model.Bom;
 import org.dependencytrack.model.Project;
 
+import java.util.UUID;
+
 public class BomProcessingFailed {
 
+    private UUID token;
     private Project project;
     private String bom;
     private String cause;
     private Bom.Format format;
     private String specVersion;
 
-    public BomProcessingFailed(final Project project, final String bom, final String cause, final Bom.Format format, final String specVersion) {
+    public BomProcessingFailed(final UUID token, final Project project, final String bom, final String cause, final Bom.Format format, final String specVersion) {
+        this.token = token;
         this.project = project;
         this.bom = bom;
         this.cause = cause;
         this.format = format;
         this.specVersion = specVersion;
     }
 
+    public UUID getToken() {
+        return token;
+    }
+
     public Project getProject() {
         return project;
     }

src/main/java/org/dependencytrack/notification/vo/ProjectVulnAnalysisComplete.java

@@ -4,18 +4,26 @@
 import org.hyades.proto.notification.v1.ProjectVulnAnalysisStatus;
 
 import java.util.List;
+import java.util.UUID;
 
 public class ProjectVulnAnalysisComplete {
+
+    private UUID token;
     private final Project project;
     private final List<ComponentVulnAnalysisComplete> findingsList;
     private final ProjectVulnAnalysisStatus status;
 
-    public ProjectVulnAnalysisComplete(Project project, List<ComponentVulnAnalysisComplete> findingsList, ProjectVulnAnalysisStatus status) {
+    public ProjectVulnAnalysisComplete(final UUID token, Project project, List<ComponentVulnAnalysisComplete> findingsList, ProjectVulnAnalysisStatus status) {
+        this.token = token;
         this.project = project;
         this.findingsList = findingsList;
         this.status = status;
     }
 
+    public UUID getToken() {
+        return token;
+    }
+
     public List<ComponentVulnAnalysisComplete> getComponentAnalysisCompleteList() {
         return findingsList;
     }

src/main/java/org/dependencytrack/parser/hyades/NotificationModelConverter.java

@@ -235,6 +235,7 @@ private static BomConsumedOrProcessedSubject convert(final BomConsumedOrProcesse
                 .build();
 
         return BomConsumedOrProcessedSubject.newBuilder()
+                .setToken(subject.getToken().toString())
                 .setProject(convert(subject.getProject()))
                 .setBom(bom)
                 .build();
@@ -248,6 +249,7 @@ private static BomProcessingFailedSubject convert(final BomProcessingFailed subj
         Optional.ofNullable(subject.getSpecVersion()).ifPresent(bomBuilder::setSpecVersion);
 
         final BomProcessingFailedSubject.Builder builder = BomProcessingFailedSubject.newBuilder()
+                .setToken(subject.getToken().toString())
                 .setProject(convert(subject.getProject()))
                 .setBom(bomBuilder.build());
 
@@ -316,6 +318,7 @@ private static ComponentVulnAnalysisCompleteSubject convert(ComponentVulnAnalysi
 
     private static ProjectVulnAnalysisCompleteSubject convert(ProjectVulnAnalysisComplete notification) {
         ProjectVulnAnalysisCompleteSubject.Builder builder = ProjectVulnAnalysisCompleteSubject.newBuilder();
+        builder.setToken(notification.getToken().toString());
         builder.setProject(convert(notification.getProject()));
         List<ComponentVulnAnalysisCompleteSubject> componentAnalysisCompleteSubjects = notification.getComponentAnalysisCompleteList().stream().map(NotificationModelConverter::convert).toList();
         builder.addAllFindings(componentAnalysisCompleteSubjects);

src/main/java/org/dependencytrack/tasks/BomUploadProcessingTask.java

@@ -148,7 +148,7 @@ public void inform(final Event e) {
                         .content("An error occurred while processing a BOM")
                         // TODO: Look into adding more fields to BomProcessingFailed, to also cover upload token, serial number, version, etc.
                         // FIXME: Add reference to BOM after we have dedicated BOM server
-                        .subject(new BomProcessingFailed(ctx.project, /* bom */ "(Omitted)", ex.getMessage(), ex.ctx.bomFormat, ex.ctx.bomSpecVersion)));
+                        .subject(new BomProcessingFailed(ctx.uploadToken, ctx.project, /* bom */ "(Omitted)", ex.getMessage(), ex.ctx.bomFormat, ex.ctx.bomSpecVersion)));
             } catch (BomProcessingException ex) {
                 LOGGER.error("BOM processing failed (%s)".formatted(ex.ctx), ex);
                 updateStateAndCancelDescendants(ctx, WorkflowStep.BOM_PROCESSING, WorkflowStatus.FAILED, ex.getMessage());
@@ -161,7 +161,7 @@ public void inform(final Event e) {
                         // TODO: Look into adding more fields to BomProcessingFailed, to also cover upload token, serial number, version, etc.
                         //   Thanks to ctx we now have more information about the BOM that may be useful to consumers downstream.
                         // FIXME: Add reference to BOM after we have dedicated BOM server
-                        .subject(new BomProcessingFailed(ctx.project, /* bom */ "(Omitted)", ex.getMessage(), ex.ctx.bomFormat, ex.ctx.bomSpecVersion)));
+                        .subject(new BomProcessingFailed(ctx.uploadToken, ctx.project, /* bom */ "(Omitted)", ex.getMessage(), ex.ctx.bomFormat, ex.ctx.bomSpecVersion)));
             } catch (Exception ex) {
                 LOGGER.error("BOM processing failed unexpectedly (%s)".formatted(ctx), ex);
                 updateStateAndCancelDescendants(ctx, WorkflowStep.BOM_PROCESSING, WorkflowStatus.FAILED, ex.getMessage());
@@ -172,7 +172,7 @@ public void inform(final Event e) {
                         .title(NotificationConstants.Title.BOM_PROCESSING_FAILED)
                         .content("An error occurred while processing a BOM")
                         // FIXME: Add reference to BOM after we have dedicated BOM server
-                        .subject(new BomProcessingFailed(ctx.project, /* bom */ "(Omitted)", ex.getMessage(), ctx.bomFormat /* (may be null) */, ctx.bomSpecVersion /* (may be null) */)));
+                        .subject(new BomProcessingFailed(ctx.uploadToken, ctx.project, /* bom */ "(Omitted)", ex.getMessage(), ctx.bomFormat /* (may be null) */, ctx.bomSpecVersion /* (may be null) */)));
             } finally {
                 timerSample.stop(TIMER);
             }
@@ -251,7 +251,7 @@ private void processBom(final Context ctx, final File bomFile) throws BomConsump
                 .level(NotificationLevel.INFORMATIONAL)
                 .title(NotificationConstants.Title.BOM_CONSUMED)
                 .content("A %s BOM was consumed and will be processed".formatted(ctx.bomFormat.getFormatShortName()))
-                .subject(new BomConsumedOrProcessed(ctx.project, /* bom */ "(Omitted)", ctx.bomFormat, ctx.bomSpecVersion)));
+                .subject(new BomConsumedOrProcessed(ctx.uploadToken, ctx.project, /* bom */ "(Omitted)", ctx.bomFormat, ctx.bomSpecVersion)));
 
 
         final var vulnAnalysisEvents = new ArrayList<ComponentVulnerabilityAnalysisEvent>();
@@ -892,7 +892,7 @@ private void dispatchBomProcessedNotification(final Context ctx) {
                 .title(NotificationConstants.Title.BOM_PROCESSED)
                 .content("A %s BOM was processed".formatted(ctx.bomFormat.getFormatShortName()))
                 // FIXME: Add reference to BOM after we have dedicated BOM server
-                .subject(new BomConsumedOrProcessed(ctx.project, /* bom */ "(Omitted)", ctx.bomFormat, ctx.bomSpecVersion)));
+                .subject(new BomConsumedOrProcessed(ctx.uploadToken, ctx.project, /* bom */ "(Omitted)", ctx.bomFormat, ctx.bomSpecVersion)));
     }
 
     /**

src/main/java/org/dependencytrack/util/NotificationUtil.java

@@ -311,7 +311,7 @@ private static void sendNotificationToKafka(UUID projectUuid, Notification notif
         new KafkaEventDispatcher().dispatchAsync(projectUuid, notification);
     }
 
-    public static Notification createProjectVulnerabilityAnalysisCompleteNotification(VulnerabilityScan vulnScan, ProjectVulnAnalysisStatus status) {
+    public static Notification createProjectVulnerabilityAnalysisCompleteNotification(VulnerabilityScan vulnScan, UUID token, ProjectVulnAnalysisStatus status) {
         try (QueryManager qm = new QueryManager()) {
             Project project = qm.getObjectByUuid(Project.class, vulnScan.getTargetIdentifier());
             if (project == null) {
@@ -360,7 +360,7 @@ public static Notification createProjectVulnerabilityAnalysisCompleteNotificatio
                     .level(NotificationLevel.INFORMATIONAL)
                     .title(NotificationConstants.Title.PROJECT_VULN_ANALYSIS_COMPLETE)
                     .content("project analysis complete for project " + project.getName() + " with id: " + project.getUuid() + " and with version: " + project.getVersion() + ". Vulnerability details added to subject ")
-                    .subject(new ProjectVulnAnalysisComplete(project, componentAnalysisCompleteList, status));
+                    .subject(new ProjectVulnAnalysisComplete(token, project, componentAnalysisCompleteList, status));
         }
     }
 

src/main/proto/org/hyades/notification/v1/notification.proto

@@ -68,12 +68,14 @@ message BackReference {
 message BomConsumedOrProcessedSubject {
   Project project = 1;
   Bom bom = 2;
+  string token = 3;
 }
 
 message BomProcessingFailedSubject {
   Project project = 1;
   Bom bom = 2;
   string cause = 3;
+  string token = 4;
 }
 
 message Bom {
@@ -208,6 +210,7 @@ message ProjectVulnAnalysisCompleteSubject {
   Project project = 1;
   repeated ComponentVulnAnalysisCompleteSubject findings = 2;
   ProjectVulnAnalysisStatus status = 3;
+  string token = 4;
 }
 
 enum ProjectVulnAnalysisStatus {

src/test/java/org/dependencytrack/parser/hyades/NotificationModelConverterTest.java

@@ -335,14 +335,14 @@ public void testConvertPolicyViolationAnalysisDecisionChangeNotification() throw
     @Test
     public void testConvertBomConsumedNotification() throws Exception {
         final org.dependencytrack.model.Project project = createProject();
-
+        final var token = UUID.randomUUID();
         final var alpineNotification = new alpine.notification.Notification();
         alpineNotification.setScope(NotificationScope.PORTFOLIO.name());
         alpineNotification.setLevel(NotificationLevel.INFORMATIONAL);
         alpineNotification.setGroup(NotificationGroup.BOM_CONSUMED.name());
         alpineNotification.setTitle("Foo");
         alpineNotification.setContent("Bar");
-        alpineNotification.setSubject(new BomConsumedOrProcessed(project, "bom", Bom.Format.CYCLONEDX, "1.4"));
+        alpineNotification.setSubject(new BomConsumedOrProcessed(token, project, "bom", Bom.Format.CYCLONEDX, "1.4"));
 
         final Notification notification = NotificationModelConverter.convert(alpineNotification);
         assertThat(notification.getScope()).isEqualTo(SCOPE_PORTFOLIO);
@@ -356,6 +356,7 @@ public void testConvertBomConsumedNotification() throws Exception {
 
         final var subject = notification.getSubject().unpack(BomConsumedOrProcessedSubject.class);
         assertProject(subject.getProject());
+        assertThat(subject.getToken()).isEqualTo(token.toString());
         assertThat(subject.getBom().getContent()).isEqualTo("bom");
         assertThat(subject.getBom().getFormat()).isEqualTo("CycloneDX");
         assertThat(subject.getBom().getSpecVersion()).isEqualTo("1.4");
@@ -364,14 +365,14 @@ public void testConvertBomConsumedNotification() throws Exception {
     @Test
     public void testConvertBomProcessedNotification() throws Exception {
         final org.dependencytrack.model.Project project = createProject();
-
+        final var token = UUID.randomUUID();
         final var alpineNotification = new alpine.notification.Notification();
         alpineNotification.setScope(NotificationScope.PORTFOLIO.name());
         alpineNotification.setLevel(NotificationLevel.INFORMATIONAL);
         alpineNotification.setGroup(NotificationGroup.BOM_PROCESSED.name());
         alpineNotification.setTitle("Foo");
         alpineNotification.setContent("Bar");
-        alpineNotification.setSubject(new BomConsumedOrProcessed(project, "bom", Bom.Format.CYCLONEDX, "1.4"));
+        alpineNotification.setSubject(new BomConsumedOrProcessed(token, project, "bom", Bom.Format.CYCLONEDX, "1.4"));
 
         final Notification notification = NotificationModelConverter.convert(alpineNotification);
         assertThat(notification.getScope()).isEqualTo(SCOPE_PORTFOLIO);
@@ -385,6 +386,7 @@ public void testConvertBomProcessedNotification() throws Exception {
 
         final var subject = notification.getSubject().unpack(BomConsumedOrProcessedSubject.class);
         assertProject(subject.getProject());
+        assertThat(subject.getToken()).isEqualTo(token.toString());
         assertThat(subject.getBom().getContent()).isEqualTo("bom");
         assertThat(subject.getBom().getFormat()).isEqualTo("CycloneDX");
         assertThat(subject.getBom().getSpecVersion()).isEqualTo("1.4");
@@ -393,14 +395,14 @@ public void testConvertBomProcessedNotification() throws Exception {
     @Test
     public void testConvertBomProcessingFailedNotification() throws Exception {
         final org.dependencytrack.model.Project project = createProject();
-
+        final var token = UUID.randomUUID();
         final var alpineNotification = new alpine.notification.Notification();
         alpineNotification.setScope(NotificationScope.PORTFOLIO.name());
         alpineNotification.setLevel(NotificationLevel.ERROR);
         alpineNotification.setGroup(NotificationGroup.BOM_PROCESSING_FAILED.name());
         alpineNotification.setTitle("Foo");
         alpineNotification.setContent("Bar");
-        alpineNotification.setSubject(new BomProcessingFailed(project, "bom", "just because", Bom.Format.CYCLONEDX, "1.4"));
+        alpineNotification.setSubject(new BomProcessingFailed(token, project, "bom", "just because", Bom.Format.CYCLONEDX, "1.4"));
 
         final Notification notification = NotificationModelConverter.convert(alpineNotification);
         assertThat(notification.getScope()).isEqualTo(SCOPE_PORTFOLIO);
@@ -414,6 +416,7 @@ public void testConvertBomProcessingFailedNotification() throws Exception {
 
         final var subject = notification.getSubject().unpack(BomProcessingFailedSubject.class);
         assertProject(subject.getProject());
+        assertThat(subject.getToken()).isEqualTo(token.toString());
         assertThat(subject.getBom().getContent()).isEqualTo("bom");
         assertThat(subject.getBom().getFormat()).isEqualTo("CycloneDX");
         assertThat(subject.getBom().getSpecVersion()).isEqualTo("1.4");
@@ -719,6 +722,7 @@ private void assertPolicyViolation(final PolicyViolation policyViolation) {
 
     @Test
     public void testConvertComponentVulnAnalysisCompleteSubject() throws Exception {
+        final var token = UUID.randomUUID();
         final org.dependencytrack.model.Project project = createProject();
         final org.dependencytrack.model.Component component = createComponent(project);
         final org.dependencytrack.model.Vulnerability vulnerability = createVulnerability();
@@ -729,7 +733,7 @@ public void testConvertComponentVulnAnalysisCompleteSubject() throws Exception {
         alpineNotification.setGroup(NotificationGroup.PROJECT_VULN_ANALYSIS_COMPLETE.name());
         alpineNotification.setTitle("Foo");
         alpineNotification.setContent("Bar");
-        alpineNotification.setSubject(new ProjectVulnAnalysisComplete(project, List.of(componentVulnAnalysisComplete), ProjectVulnAnalysisStatus.PROJECT_VULN_ANALYSIS_STATUS_COMPLETED));
+        alpineNotification.setSubject(new ProjectVulnAnalysisComplete(token, project, List.of(componentVulnAnalysisComplete), ProjectVulnAnalysisStatus.PROJECT_VULN_ANALYSIS_STATUS_COMPLETED));
 
         final Notification notification = NotificationModelConverter.convert(alpineNotification);
         assertThat(notification.getScope()).isEqualTo(SCOPE_PORTFOLIO);
@@ -743,6 +747,7 @@ public void testConvertComponentVulnAnalysisCompleteSubject() throws Exception {
 
         final var subject = notification.getSubject().unpack(ProjectVulnAnalysisCompleteSubject.class);
         assertProject(subject.getProject());
+        assertThat(subject.getToken()).isEqualTo(token.toString());
         assertComponent(subject.getFindingsList().get(0).getComponent());
         assertVulnerability(subject.getFindingsList().get(0).getVulnerabilities(0));
         assertThat(subject.getStatus()).isEqualTo(ProjectVulnAnalysisStatus.PROJECT_VULN_ANALYSIS_STATUS_COMPLETED);