Convert CSAF tasks to workflows

api/src/main/openapi/paths/csaf/aggregators__id__provider-discovery.yaml

@@ -42,5 +42,7 @@ post:
       $ref: "../../components/responses/generic-forbidden-error.yaml"
     "404":
       $ref: "../../components/responses/generic-not-found-error.yaml"
+    "409":
+      $ref: "../../components/responses/generic-conflict-error.yaml"
     default:
       $ref: "../../components/responses/generic-error.yaml"

apiserver/src/main/java/org/dependencytrack/csaf/CsafProviderDiscoveryTask.java → apiserver/src/main/java/org/dependencytrack/csaf/DiscoverCsafProvidersActivity.java

@@ -18,73 +18,72 @@
  */
 package org.dependencytrack.csaf;
 
-import alpine.event.framework.Event;
-import alpine.event.framework.Subscriber;
+import org.dependencytrack.dex.api.Activity;
+import org.dependencytrack.dex.api.ActivityContext;
+import org.dependencytrack.dex.api.ActivitySpec;
+import org.dependencytrack.dex.api.failure.TerminalApplicationFailureException;
+import org.dependencytrack.proto.internal.workflow.v1.DiscoverCsafProvidersArg;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
 
 import java.time.Instant;
 import java.util.List;
-import java.util.concurrent.ExecutionException;
+import java.util.UUID;
 
 import static org.dependencytrack.persistence.jdbi.JdbiFactory.inJdbiTransaction;
 import static org.dependencytrack.persistence.jdbi.JdbiFactory.useJdbiTransaction;
+import static org.dependencytrack.persistence.jdbi.JdbiFactory.withJdbiHandle;
 
 /**
- * TODO: Refactor to dex workflow + activity once dex engine is integrated.
- *   * Use workflow instance ID to ensure only one workflow run per aggregator can exist.
- *   * Use same concurrency key across all aggregators to serialize their execution.
- *
  * @since 5.7.0
  */
-public class CsafProviderDiscoveryTask implements Subscriber {
+@ActivitySpec(name = "discover-csaf-providers")
+public final class DiscoverCsafProvidersActivity implements Activity<DiscoverCsafProvidersArg, Void> {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(CsafProviderDiscoveryTask.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(DiscoverCsafProvidersActivity.class);
 
     private final CsafClient csafClient;
 
-    CsafProviderDiscoveryTask(CsafClient csafClient) {
+    DiscoverCsafProvidersActivity(CsafClient csafClient) {
         this.csafClient = csafClient;
     }
 
-    @SuppressWarnings("unused") // Used by event system.
-    public CsafProviderDiscoveryTask() {
+    public DiscoverCsafProvidersActivity() {
         this(new CsafClient());
     }
 
     @Override
-    public void inform(Event e) {
-        if (!(e instanceof final CsafProviderDiscoveryEvent event)) {
-            return;
+    public @Nullable Void execute(
+            ActivityContext ctx,
+            @Nullable DiscoverCsafProvidersArg arg) throws Exception {
+        if (arg == null) {
+            throw new TerminalApplicationFailureException("No argument provided");
         }
 
-        final CsafAggregator aggregator = event.getAggregator();
+        final CsafAggregator aggregator = withJdbiHandle(
+                handle -> handle.attach(CsafAggregatorDao.class).getById(UUID.fromString(arg.getAggregatorId())));
+        if (aggregator == null) {
+            throw new TerminalApplicationFailureException(
+                    "Aggregator with ID %s does not exist".formatted(arg.getAggregatorId()));
+        }
 
         try (var ignored = MDC.putCloseable("csafAggregator", aggregator.getNamespace().toString())) {
             LOGGER.info("Discovering providers");
 
-            final List<CsafProvider> discoveredProviders;
-            try {
-                discoveredProviders =
-                        csafClient.discoverProviders(aggregator)
-                                .peek(provider -> {
-                                    provider.setEnabled(false);
-                                    provider.setDiscoveredFrom(aggregator.getId());
-                                    provider.setDiscoveredAt(Instant.now());
-                                })
-                                .toList();
-            } catch (ExecutionException | RuntimeException ex) {
-                throw new IllegalStateException("Failed to discover providers", ex);
-            } catch (InterruptedException ex) {
-                LOGGER.warn("Interrupted while discovering providers", ex);
-                Thread.currentThread().interrupt();
-                return;
-            }
+            final List<CsafProvider> discoveredProviders =
+                    csafClient.discoverProviders(aggregator)
+                            .peek(provider -> {
+                                provider.setEnabled(false);
+                                provider.setDiscoveredFrom(aggregator.getId());
+                                provider.setDiscoveredAt(Instant.now());
+                            })
+                            .toList();
 
             if (discoveredProviders.isEmpty()) {
                 LOGGER.info("No providers discovered");
-                return;
+                return null;
             }
 
             LOGGER.info("Discovered {} providers", discoveredProviders.size());
@@ -98,6 +97,8 @@ public void inform(Event e) {
             useJdbiTransaction(handle -> handle.attach(CsafAggregatorDao.class)
                     .updateLastDiscoveryAtById(aggregator.getId(), Instant.now()));
         }
+
+        return null;
     }
 
 }

apiserver/src/main/java/org/dependencytrack/csaf/DiscoverCsafProvidersWorkflow.java

@@ -0,0 +1,46 @@
+/*
+ * This file is part of Dependency-Track.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *   http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ *
+ * SPDX-License-Identifier: Apache-2.0
+ * Copyright (c) OWASP Foundation. All Rights Reserved.
+ */
+package org.dependencytrack.csaf;
+
+import org.dependencytrack.dex.api.Workflow;
+import org.dependencytrack.dex.api.WorkflowContext;
+import org.dependencytrack.dex.api.WorkflowSpec;
+import org.dependencytrack.dex.api.failure.TerminalApplicationFailureException;
+import org.dependencytrack.proto.internal.workflow.v1.DiscoverCsafProvidersArg;
+import org.jspecify.annotations.Nullable;
+
+/**
+ * @since 5.7.0
+ */
+@WorkflowSpec(name = "discover-csaf-providers")
+public final class DiscoverCsafProvidersWorkflow implements Workflow<DiscoverCsafProvidersArg, Void> {
+
+    @Override
+    public @Nullable Void execute(
+            WorkflowContext<DiscoverCsafProvidersArg> ctx,
+            @Nullable DiscoverCsafProvidersArg arg) throws Exception {
+        if (arg == null) {
+            throw new TerminalApplicationFailureException("No argument provided");
+        }
+
+        ctx.activity(DiscoverCsafProvidersActivity.class).call(arg).await();
+        return null;
+    }
+
+}

apiserver/src/main/java/org/dependencytrack/csaf/CsafDocumentImportTask.java → apiserver/src/main/java/org/dependencytrack/csaf/ImportCsafDocumentsActivity.java

@@ -18,14 +18,17 @@
  */
 package org.dependencytrack.csaf;
 
-import alpine.event.framework.Event;
-import alpine.event.framework.Subscriber;
 import io.csaf.retrieval.RetrievedDocument;
-import org.dependencytrack.common.pagination.PageIterator;
+import org.dependencytrack.dex.api.Activity;
+import org.dependencytrack.dex.api.ActivityContext;
+import org.dependencytrack.dex.api.ActivitySpec;
+import org.dependencytrack.dex.api.failure.TerminalApplicationFailureException;
 import org.dependencytrack.model.Advisory;
 import org.dependencytrack.model.Vulnerability;
 import org.dependencytrack.persistence.QueryManager;
+import org.dependencytrack.proto.internal.workflow.v1.ImportCsafDocumentsArg;
 import org.jdbi.v3.core.Handle;
+import org.jspecify.annotations.Nullable;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.MDC;
@@ -35,6 +38,7 @@
 import java.util.HashSet;
 import java.util.Iterator;
 import java.util.List;
+import java.util.UUID;
 import java.util.concurrent.ExecutionException;
 import java.util.stream.Stream;
 
@@ -43,68 +47,53 @@
 import static org.dependencytrack.persistence.jdbi.JdbiFactory.withJdbiHandle;
 
 /**
- * TODO: Refactor to dex workflow + activity once dex engine is integrated.
- *   * Use workflow instance ID to ensure only one workflow run per provider can exist.
- *   * Use same concurrency key across all providers to serialize their execution.
- *   * Create an "uber-workflow" that that triggers import for all providers.
- *   * Schedule the uber-workflow to run at least once daily.
- *
  * @since 5.7.0
  */
-public class CsafDocumentImportTask implements Subscriber {
+@ActivitySpec(name = "import-csaf-documents")
+public final class ImportCsafDocumentsActivity implements Activity<ImportCsafDocumentsArg, Void> {
 
-    private static final Logger LOGGER = LoggerFactory.getLogger(CsafDocumentImportTask.class);
+    private static final Logger LOGGER = LoggerFactory.getLogger(ImportCsafDocumentsActivity.class);
 
     private final CsafClient csafClient;
 
-    CsafDocumentImportTask(CsafClient csafClient) {
+    ImportCsafDocumentsActivity(CsafClient csafClient) {
         this.csafClient = csafClient;
     }
 
-    @SuppressWarnings("unused")
-    public CsafDocumentImportTask() {
+    public ImportCsafDocumentsActivity() {
         this(new CsafClient());
     }
 
     @Override
-    public void inform(Event e) {
-        if (!(e instanceof CsafDocumentImportEvent)) {
-            return;
+    public @Nullable Void execute(
+            ActivityContext ctx,
+            @Nullable ImportCsafDocumentsArg arg) throws Exception {
+        if (arg == null) {
+            throw new TerminalApplicationFailureException("No argument provided");
         }
 
-        final List<CsafProvider> providers = getEnabledProviders();
-        if (providers.isEmpty()) {
-            LOGGER.info("No providers available to import documents from");
-            return;
+        final CsafProvider provider = withJdbiHandle(
+                handle -> handle.attach(CsafProviderDao.class).getById(UUID.fromString(arg.getProviderId())));
+        if (provider == null) {
+            throw new TerminalApplicationFailureException(
+                    "Provider with ID %s does not exist".formatted(arg.getProviderId()));
         }
 
-        for (final CsafProvider provider : providers) {
-            try (var ignored = MDC.putCloseable("csafProvider", provider.getName())) {
-                importDocuments(provider);
-            } catch (ExecutionException | RuntimeException ex) {
-                LOGGER.error("Failed to import CSAF documents", ex);
-            } catch (InterruptedException ex) {
-                LOGGER.warn("Interrupted while importing CSAF documents");
-                Thread.currentThread().interrupt();
-                break;
+        try (var ignored = MDC.putCloseable("csafProvider", provider.getName())) {
+            if (!provider.isEnabled()) {
+                LOGGER.info("Provider is disabled");
+                return null;
             }
+
+            importDocuments(ctx, provider);
         }
-    }
 
-    private List<CsafProvider> getEnabledProviders() {
-        return withJdbiHandle(handle -> {
-            final var dao = handle.attach(CsafProviderDao.class);
-
-            return PageIterator.stream(
-                            pageToken -> dao.list(
-                                    new ListCsafProvidersQuery()
-                                            .withEnabled(true)
-                                            .withPageToken(pageToken)))
-                    .toList();
-        });
+        return null;
     }
 
-    private void importDocuments(CsafProvider provider) throws ExecutionException, InterruptedException {
+    private void importDocuments(
+            ActivityContext ctx,
+            CsafProvider provider) throws ExecutionException, InterruptedException {
         Instant latestDocumentReleaseDate = provider.getLatestDocumentReleaseDate();
         if (latestDocumentReleaseDate != null) {
             LOGGER.info("Importing CSAF documents modified since {}", latestDocumentReleaseDate);
@@ -119,6 +108,7 @@ private void importDocuments(CsafProvider provider) throws ExecutionException, I
             final var documentBatch = new ArrayList<RetrievedDocument>(25);
             while (documentIterator.hasNext()) {
                 final RetrievedDocument doc = documentIterator.next();
+                ctx.maybeHeartbeat();
 
                 final Instant docReleaseDate = doc.getJson().getDocument().getTracking().getCurrent_release_date().getValue$kotlinx_datetime();
                 if (latestDocumentReleaseDate == null || latestDocumentReleaseDate.isBefore(docReleaseDate)) {
@@ -133,6 +123,7 @@ private void importDocuments(CsafProvider provider) throws ExecutionException, I
             }
 
             if (!documentBatch.isEmpty()) {
+                ctx.maybeHeartbeat();
                 processDocuments(documentBatch);
                 documentBatch.clear();
             }