Support multiple configurable OSV data sources

migration/src/main/resources/org/dependencytrack/migration/V202608241029__osv_multi_source_config.sql

@@ -0,0 +1,19 @@
+-- Convert legacy OSV config to multi-source config
+-- example: {"enabled": true, ...} -> {"sources": [{"name":"default", "enabled": true, ...}]}.
+UPDATE "EXTENSION_RUNTIME_CONFIG"
+    SET "CONFIG" = jsonb_build_object(
+        'sources',
+        jsonb_build_array( "CONFIG" || jsonb_build_object('name', 'default') )
+    ),
+    "UPDATED_AT" = now()
+WHERE "EXTENSION_POINT" = 'vuln-data-source'
+    AND "EXTENSION" = 'osv'
+    AND NOT ("CONFIG" ? 'sources');
+
+-- Watermarks are namespaced by source name. Move existing watermarks to the default source.
+UPDATE "EXTENSION_KV_STORE"
+    SET "KEY" = 'watermark/default/' || replace("KEY", 'watermark/', '')
+WHERE "EXTENSION_POINT" = 'vuln-data-source'
+    AND "EXTENSION" = 'osv'
+    AND "KEY" LIKE 'watermark/%'
+    AND NOT "KEY" LIKE 'watermark/default/%';

vuln-data-source/osv/src/main/java/org/dependencytrack/vulndatasource/osv/OsvCompositeVulnDataSource.java

@@ -0,0 +1,106 @@
+/*
+ * 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.vulndatasource.osv;
+
+import org.cyclonedx.proto.v1_7.Bom;
+import org.dependencytrack.vulndatasource.api.VulnDataSource;
+import org.jspecify.annotations.Nullable;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import org.slf4j.MDC;
+
+import java.util.Collections;
+import java.util.IdentityHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.NoSuchElementException;
+
+import static java.util.Objects.requireNonNull;
+
+/**
+ * @since 5.0.0
+ */
+final class OsvCompositeVulnDataSource implements VulnDataSource {
+
+    private static final Logger LOGGER = LoggerFactory.getLogger(OsvCompositeVulnDataSource.class);
+    private final List<OsvVulnDataSource> dataSources;
+    private @Nullable OsvVulnDataSource currentDataSource;
+    private int currentDataSourceIndex;
+
+    /**
+     * Tracks the originating data source for a Bom instance so markProcessed can be
+     * delegated to the producer even if currentDataSource has moved on.
+     */
+    private final Map<Bom, OsvVulnDataSource> originMap = Collections.synchronizedMap(new IdentityHashMap<>());
+
+    OsvCompositeVulnDataSource(final List<OsvVulnDataSource> dataSources) {
+        this.dataSources = requireNonNull(dataSources, "dataSources must not be null");
+    }
+
+    @Override
+    public boolean hasNext() {
+        while (currentDataSourceIndex < dataSources.size()) {
+            if (dataSources.get(currentDataSourceIndex).hasNext()) {
+                return true;
+            }
+            currentDataSourceIndex++;
+        }
+        return false;
+    }
+
+    @Override
+    public Bom next() {
+        if (currentDataSourceIndex >= dataSources.size()) {
+            throw new NoSuchElementException();
+        }
+        currentDataSource = dataSources.get(currentDataSourceIndex);
+        try (final var _ = MDC.putCloseable("osvSource", currentDataSource.getDataSourceName())) {
+            final Bom bom = currentDataSource.next();
+            originMap.put(bom, currentDataSource);
+            return bom;
+        }
+    }
+
+    @Override
+    public void markProcessed(final Bom bom) {
+        final var origin = originMap.remove(bom);
+        final var target = origin != null ? origin : currentDataSource;
+        if (target == null) {
+            throw new IllegalStateException("No data source available to mark processed");
+        }
+        try (final var _ = MDC.putCloseable("osvSource", target.getDataSourceName())) {
+            target.markProcessed(bom);
+        }
+    }
+
+    @Override
+    public void close() {
+        for (final var dataSource : dataSources) {
+            try {
+                dataSource.close();
+            } catch (final Exception e) {
+                LOGGER.warn("Failed to close data source: {}", dataSource.getDataSourceName(), e);
+            }
+        }
+    }
+
+    List<OsvVulnDataSource> getDataSources() {
+        return dataSources;
+    }
+}

vuln-data-source/osv/src/main/java/org/dependencytrack/vulndatasource/osv/OsvVulnDataSource.java

@@ -62,6 +62,7 @@ final class OsvVulnDataSource implements VulnDataSource {
     private static final Logger LOGGER = LoggerFactory.getLogger(OsvVulnDataSource.class);
     private static final int MAX_INCREMENTAL_ADVISORY_DOWNLOADS = 250;
 
+    private final String dataSourceName;
     private final @Nullable WatermarkManager watermarkManager;
     private final ObjectMapper objectMapper;
     private final String dataUrl;
@@ -78,12 +79,14 @@ final class OsvVulnDataSource implements VulnDataSource {
     private final boolean isAliasSyncEnabled;
 
     OsvVulnDataSource(
+            final String dataSourceName,
             final @Nullable WatermarkManager watermarkManager,
             final ObjectMapper objectMapper,
             final String dataUrl,
             final Collection<String> ecosystems,
             final HttpClient httpClient,
             final boolean isAliasSyncEnabled) {
+        this.dataSourceName = requireNonNull(dataSourceName, "dataSourceName must not be null");
         this.watermarkManager = watermarkManager;
         this.objectMapper = objectMapper;
         this.dataUrl = dataUrl;
@@ -213,8 +216,9 @@ private void logCurrentEcosystemSummary() {
         }
 
         LOGGER.info(
-                "Finished ecosystem {}: processed {} advisories",
+                "Finished ecosystem {} of data source {}: processed {} advisories",
                 currentEcosystem,
+                dataSourceName,
                 currentEcosystemAdvisoriesProcessed);
     }
 
@@ -223,7 +227,7 @@ private void openNextEcosystem() {
         currentEcosystemAdvisoriesProcessed = 0;
         currentAdvisorySource = openAdvisorySource(currentEcosystem);
 
-        LOGGER.info("Processing ecosystem {}", currentEcosystem);
+        LOGGER.info("Processing ecosystem {} of data source {}", currentEcosystem, dataSourceName);
     }
 
     private @Nullable OsvAdvisorySource openAdvisorySource(String ecosystem) {
@@ -376,4 +380,8 @@ private Set<String> getModifiedAdvisoryIds(String ecosystem, Instant watermark)
     WatermarkManager getWatermarkManager() {
         return watermarkManager;
     }
+
+    String getDataSourceName() {
+        return dataSourceName;
+    }
 }

vuln-data-source/osv/src/main/java/org/dependencytrack/vulndatasource/osv/OsvVulnDataSourceFactory.java

@@ -32,6 +32,9 @@
 
 import java.net.URI;
 import java.net.http.HttpClient;
+import java.util.ArrayList;
+import java.util.LinkedHashSet;
+import java.util.List;
 import java.util.Set;
 
 import static java.util.Objects.requireNonNull;
@@ -41,6 +44,7 @@
  */
 final class OsvVulnDataSourceFactory implements VulnDataSourceFactory, RuntimeConfigurable {
 
+    private static final String DEFAULT_SOURCE_NAME = "default";
     private @Nullable ConfigRegistry configRegistry;
     private @Nullable KeyValueStore kvStore;
     private @Nullable ObjectMapper objectMapper;
@@ -76,30 +80,43 @@ public void init(ExtensionContext context) {
 
     @Override
     public RuntimeConfigSpec runtimeConfigSpec() {
-        final var defaultConfig = new OsvVulnDataSourceConfigV1()
+        final var defaultSource = new OsvSourceConfigV1()
+                .withName(DEFAULT_SOURCE_NAME)
                 .withIncrementalMirroringEnabled(true)
                 .withEnabled(false)
                 .withAliasSyncEnabled(false)
                 .withDataUrl(URI.create("https://storage.googleapis.com/osv-vulnerabilities"))
                 .withEcosystems(Set.of("Go", "Maven", "npm", "NuGet", "PyPI"));
 
-        return RuntimeConfigSpec.of(defaultConfig, config -> {
-            if (!config.isEnabled()) {
-                return;
-            }
-            if (config.getDataUrl() == null) {
-                throw new InvalidRuntimeConfigException("No data URL provided");
-            }
-            if (config.getEcosystems() == null || config.getEcosystems().isEmpty()) {
-                throw new InvalidRuntimeConfigException("At least one ecosystem must be specified");
+        final var defaultConfig = new OsvVulnDataSourceConfigV1().withFeeds(new LinkedHashSet<>(Set.of(defaultSource)));
+
+        return RuntimeConfigSpec.of(defaultConfig, (OsvVulnDataSourceConfigV1 config) -> {
+            final Set<String> seenNames = new LinkedHashSet<>();
+            for (final var feed : config.getFeeds()) {
+                if (feed.getName() == null || feed.getName().isBlank()) {
+                    throw new InvalidRuntimeConfigException("No data feed name provided");
+                }
+                if (!seenNames.add(feed.getName())) {
+                    throw new InvalidRuntimeConfigException("Duplicate data feed name provided: " + feed.getName());
+                }
+                if (!feed.isEnabled()) {
+                    continue;
+                }
+                if (feed.getDataUrl() == null) {
+                    throw new InvalidRuntimeConfigException("No data URL provided");
+                }
+                if (feed.getEcosystems() == null || feed.getEcosystems().isEmpty()) {
+                    throw new InvalidRuntimeConfigException("At least one ecosystem must be specified");
+                }
             }
         });
     }
 
     @Override
     public boolean isDataSourceEnabled() {
         requireNonNull(configRegistry, "configRegistry must not be null");
-        return configRegistry.getRuntimeConfig(OsvVulnDataSourceConfigV1.class).isEnabled();
+        return !enabledFeeds(configRegistry.getRuntimeConfig(OsvVulnDataSourceConfigV1.class))
+                .isEmpty();
     }
 
     @Override
@@ -109,20 +126,31 @@ public VulnDataSource create() {
         requireNonNull(objectMapper, "objectMapper must not be null");
         requireNonNull(httpClient, "httpClient must not be null");
 
-        final var config = configRegistry.getRuntimeConfig(OsvVulnDataSourceConfigV1.class);
-        if (!config.isEnabled()) {
+        final List<OsvSourceConfigV1> feeds =
+                enabledFeeds(configRegistry.getRuntimeConfig(OsvVulnDataSourceConfigV1.class));
+        if (feeds.isEmpty()) {
             throw new IllegalStateException("Vulnerability data source is disabled and cannot be created");
         }
 
-        final WatermarkManager watermarkManager =
-                config.isIncrementalMirroringEnabled() ? new WatermarkManager(config.getEcosystems(), kvStore) : null;
+        final var dataSources = new ArrayList<OsvVulnDataSource>(feeds.size());
+        for (final OsvSourceConfigV1 feed : feeds) {
+            final WatermarkManager watermarkManager = feed.isIncrementalMirroringEnabled()
+                    ? new WatermarkManager(feed.getName(), feed.getEcosystems(), kvStore)
+                    : null;
+
+            dataSources.add(new OsvVulnDataSource(
+                    feed.getName(),
+                    watermarkManager,
+                    objectMapper,
+                    feed.getDataUrl().toString(),
+                    feed.getEcosystems(),
+                    httpClient,
+                    feed.getAliasSyncEnabled()));
+        }
+        return new OsvCompositeVulnDataSource(dataSources);
+    }
 
-        return new OsvVulnDataSource(
-                watermarkManager,
-                objectMapper,
-                config.getDataUrl().toString(),
-                config.getEcosystems(),
-                httpClient,
-                config.getAliasSyncEnabled());
+    private List<OsvSourceConfigV1> enabledFeeds(final OsvVulnDataSourceConfigV1 config) {
+        return config.getFeeds().stream().filter(OsvSourceConfigV1::isEnabled).toList();
     }
 }

vuln-data-source/osv/src/main/java/org/dependencytrack/vulndatasource/osv/WatermarkManager.java

@@ -39,8 +39,8 @@ final class WatermarkManager {
     private final Map<String, WatermarkRecord> pendingRecordByEcosystem;
     private final Map<String, WatermarkRecord> committedRecordByEcosystem;
 
-    WatermarkManager(final Collection<String> ecosystems, final KeyValueStore kvStore) {
-        final var watermarkStore = new WatermarkStore(kvStore);
+    WatermarkManager(final String sourceName, final Collection<String> ecosystems, final KeyValueStore kvStore) {
+        final var watermarkStore = new WatermarkStore(sourceName, kvStore);
         final Map<String, WatermarkRecord> recordByEcosystem = watermarkStore.getForEcosystems(ecosystems);
 
         this.store = watermarkStore;

vuln-data-source/osv/src/main/java/org/dependencytrack/vulndatasource/osv/WatermarkStore.java

@@ -40,15 +40,17 @@ final class WatermarkStore {
 
     private static final Logger LOGGER = LoggerFactory.getLogger(WatermarkStore.class);
 
+    private final String sourceName;
     private final KeyValueStore kvStore;
 
-    WatermarkStore(final KeyValueStore kvStore) {
+    WatermarkStore(final String sourceName, final KeyValueStore kvStore) {
+        this.sourceName = requireNonNull(sourceName, "OSV sourceName must not be null");
         this.kvStore = kvStore;
     }
 
     Map<String, WatermarkRecord> getForEcosystems(final Collection<String> ecosystems) {
         final Map<String, String> ecosystemByKey =
-                ecosystems.stream().collect(Collectors.toMap(WatermarkStore::getKey, Function.identity()));
+                ecosystems.stream().collect(Collectors.toMap(this::getKey, Function.identity()));
 
         final Map<String, KeyValueStore.Entry> kvEntryByKey = kvStore.getMany(ecosystemByKey.keySet());
         if (kvEntryByKey.isEmpty()) {
@@ -92,7 +94,7 @@ WatermarkRecord save(final WatermarkRecord watermark) {
         };
     }
 
-    private static String getKey(final String ecosystem) {
-        return "watermark/" + ecosystem;
+    private String getKey(final String ecosystem) {
+        return "watermark/" + sourceName + "/" + ecosystem;
     }
 }