Merge pull request #1443 from DependencyTrack/migrate-epss-mirroring-

Author: nscuro

apiserver/src/main/java/org/dependencytrack/dev/DevServicesInitializer.java

@@ -19,14 +19,14 @@
 package org.dependencytrack.dev;
 
 import alpine.common.logging.Logger;
+import jakarta.servlet.ServletContextEvent;
+import jakarta.servlet.ServletContextListener;
 import org.apache.kafka.clients.admin.AdminClient;
 import org.apache.kafka.clients.admin.NewTopic;
 import org.dependencytrack.event.kafka.KafkaTopics;
 import org.eclipse.microprofile.config.Config;
 import org.eclipse.microprofile.config.ConfigProvider;
 
-import jakarta.servlet.ServletContextEvent;
-import jakarta.servlet.ServletContextListener;
 import java.lang.reflect.Constructor;
 import java.lang.reflect.Method;
 import java.util.ArrayList;
@@ -166,7 +166,6 @@ public void contextInitialized(final ServletContextEvent event) {
         }
 
         final var topicsToCreate = new ArrayList<>(List.of(
-                new NewTopic(KafkaTopics.NEW_EPSS.name(), 1, (short) 1).configs(Map.of(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_COMPACT)),
                 new NewTopic(KafkaTopics.NEW_VULNERABILITY.name(), 1, (short) 1).configs(Map.of(CLEANUP_POLICY_CONFIG, CLEANUP_POLICY_COMPACT)),
                 new NewTopic(KafkaTopics.NOTIFICATION_ANALYZER.name(), 1, (short) 1),
                 new NewTopic(KafkaTopics.NOTIFICATION_BOM.name(), 1, (short) 1),

apiserver/src/main/java/org/dependencytrack/event/kafka/KafkaEventConverter.java

@@ -23,7 +23,6 @@
 import com.google.protobuf.Message;
 import org.dependencytrack.event.ComponentRepositoryMetaAnalysisEvent;
 import org.dependencytrack.event.ComponentVulnerabilityAnalysisEvent;
-import org.dependencytrack.event.EpssMirrorEvent;
 import org.dependencytrack.event.OsvMirrorEvent;
 import org.dependencytrack.event.kafka.KafkaTopics.Topic;
 import org.dependencytrack.model.Vulnerability;
@@ -69,7 +68,6 @@ private KafkaEventConverter() {
             case ComponentRepositoryMetaAnalysisEvent e -> convert(e);
             case ComponentVulnerabilityAnalysisEvent e -> convert(e);
             case OsvMirrorEvent e -> convert(e);
-            case EpssMirrorEvent e -> convert(e);
             default -> throw new IllegalArgumentException("Unable to convert event " + event);
         };
     }
@@ -150,10 +148,6 @@ static KafkaEvent<String, String> convert(final OsvMirrorEvent event) {
         return new KafkaEvent<>(KafkaTopics.VULNERABILITY_MIRROR_COMMAND, key, value);
     }
 
-    static KafkaEvent<String, String> convert(final EpssMirrorEvent ignored) {
-        return new KafkaEvent<>(KafkaTopics.VULNERABILITY_MIRROR_COMMAND, "EPSS", null);
-    }
-
     private static Topic<String, Notification> extractDestinationTopic(final Notification notification) {
         return switch (notification.getGroup()) {
             case GROUP_ANALYZER -> KafkaTopics.NOTIFICATION_ANALYZER;

apiserver/src/main/java/org/dependencytrack/event/kafka/KafkaTopics.java

@@ -24,7 +24,6 @@
 import org.cyclonedx.proto.v1_6.Bom;
 import org.dependencytrack.common.ConfigKey;
 import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerde;
-import org.dependencytrack.proto.mirror.v1.EpssItem;
 import org.dependencytrack.proto.notification.v1.Notification;
 import org.dependencytrack.proto.repometaanalysis.v1.AnalysisCommand;
 import org.dependencytrack.proto.repometaanalysis.v1.AnalysisResult;
@@ -57,7 +56,6 @@ public final class KafkaTopics {
     public static final Topic<String, ScanResult> VULN_ANALYSIS_RESULT_PROCESSED;
 
     public static final Topic<String, Notification> NOTIFICATION_PROJECT_VULN_ANALYSIS_COMPLETE;
-    public static final Topic<String, EpssItem> NEW_EPSS;
     private static final Serde<Notification> NOTIFICATION_SERDE = new KafkaProtobufSerde<>(Notification.parser());
 
     static {
@@ -82,7 +80,6 @@ public final class KafkaTopics {
         VULN_ANALYSIS_RESULT = new Topic<>("dtrack.vuln-analysis.result", new KafkaProtobufSerde<>(ScanKey.parser()), new KafkaProtobufSerde<>(ScanResult.parser()));
         VULN_ANALYSIS_RESULT_PROCESSED = new Topic<>("dtrack.vuln-analysis.result.processed", Serdes.String(), new KafkaProtobufSerde<>(ScanResult.parser()));
         NOTIFICATION_PROJECT_VULN_ANALYSIS_COMPLETE = new Topic<>("dtrack.notification.project-vuln-analysis-complete", Serdes.String(), NOTIFICATION_SERDE);
-        NEW_EPSS = new Topic<>("dtrack.epss", Serdes.String(), new KafkaProtobufSerde<>(EpssItem.parser()));
         NOTIFICATION_USER = new Topic<>("dtrack.notification.user", Serdes.String(), NOTIFICATION_SERDE);
     }
 

apiserver/src/main/java/org/dependencytrack/event/kafka/processor/EpssMirrorProcessor.java

@@ -19,11 +19,10 @@
 package org.dependencytrack.event.kafka.processor;
 
 import alpine.common.logging.Logger;
-import org.dependencytrack.event.kafka.KafkaTopics;
-import org.dependencytrack.event.kafka.processor.api.ProcessorManager;
-
 import jakarta.servlet.ServletContextEvent;
 import jakarta.servlet.ServletContextListener;
+import org.dependencytrack.event.kafka.KafkaTopics;
+import org.dependencytrack.event.kafka.processor.api.ProcessorManager;
 
 public class ProcessorInitializer implements ServletContextListener {
 
@@ -39,8 +38,6 @@ public void contextInitialized(final ServletContextEvent event) {
                 KafkaTopics.NEW_VULNERABILITY, new VulnerabilityMirrorProcessor());
         PROCESSOR_MANAGER.registerProcessor(RepositoryMetaResultProcessor.PROCESSOR_NAME,
                 KafkaTopics.REPO_META_ANALYSIS_RESULT, new RepositoryMetaResultProcessor());
-        PROCESSOR_MANAGER.registerBatchProcessor(EpssMirrorProcessor.PROCESSOR_NAME,
-                KafkaTopics.NEW_EPSS, new EpssMirrorProcessor());
         PROCESSOR_MANAGER.registerProcessor(VulnerabilityScanResultProcessor.PROCESSOR_NAME,
                 KafkaTopics.VULN_ANALYSIS_RESULT, new VulnerabilityScanResultProcessor());
         PROCESSOR_MANAGER.registerBatchProcessor(ProcessedVulnerabilityScanResultProcessor.PROCESSOR_NAME,

apiserver/src/main/java/org/dependencytrack/event/kafka/processor/ProcessorInitializer.java

@@ -30,6 +30,8 @@
 import java.io.Serializable;
 import java.math.BigDecimal;
 
+import static java.util.Objects.requireNonNull;
+
 @PersistenceCapable
 @JsonInclude(JsonInclude.Include.NON_NULL)
 public class Epss implements Serializable {
@@ -52,6 +54,15 @@ public class Epss implements Serializable {
     @Column(name = "PERCENTILE", scale = 5)
     private BigDecimal percentile;
 
+    public Epss() {
+    }
+
+    public Epss(final String cve, final BigDecimal score, final BigDecimal percentile) {
+        this.cve = requireNonNull(cve, "cve must not be null");
+        this.score = requireNonNull(score, "score must not be null");
+        this.percentile = requireNonNull(percentile, "percentile must not be null");
+    }
+
     public long getId() {
         return id;
     }

apiserver/src/main/java/org/dependencytrack/model/Epss.java

@@ -27,8 +27,6 @@
 import java.util.function.Function;
 import java.util.stream.Collectors;
 
-import static org.dependencytrack.util.PersistenceUtil.applyIfChanged;
-
 final class EpssQueryManager extends QueryManager implements IQueryManager {
 
     /**
@@ -40,44 +38,6 @@ final class EpssQueryManager extends QueryManager implements IQueryManager {
         super(pm);
     }
 
-    /**
-     * Synchronizes a Epss record. Method first checkes to see if the record already
-     * exists and if so, updates it. If the record does not already exist,
-     * this method will create a new Epss record.
-     * @param epss the Epss record to synchronize
-     * @return a Epss object
-     */
-    public Epss synchronizeEpss(Epss epss) {
-        Epss result = updateEpss(epss);
-        if (result == null) {
-            final Epss epssNew = pm.makePersistent(epss);
-            return epssNew;
-        }
-        return result;
-    }
-
-    /**
-     * Synchronizes a batch of Epss records.
-     * @param epssList the batch of Epss records to synchronize
-     */
-    public void synchronizeAllEpss(List<Epss> epssList) {
-        runInTransaction(() -> {
-            for (final Epss epss : epssList) {
-                synchronizeEpss(epss);
-            }
-        });
-    }
-
-    private Epss updateEpss(Epss epss) {
-        var epssExisting = getEpssByCveId(epss.getCve());
-        if (epssExisting != null) {
-            applyIfChanged(epssExisting, epss, Epss::getScore, epssExisting::setScore);
-            applyIfChanged(epssExisting, epss, Epss::getPercentile, epssExisting::setPercentile);
-            return epssExisting;
-        }
-        return null;
-    }
-
     /**
      * Returns a Epss record by its CVE id.
      * @param cveId the CVE id of the record

apiserver/src/main/java/org/dependencytrack/parser/dependencytrack/EpssModelConverter.java

@@ -1613,14 +1613,6 @@ public VulnerabilityPolicyBundle getVulnerabilityPolicyBundle() {
         return singleResult(query.execute());
     }
 
-    public Epss synchronizeEpss(Epss epss) {
-        return getEpssQueryManager().synchronizeEpss(epss);
-    }
-
-    public void synchronizeAllEpss(List<Epss> epssList) {
-        getEpssQueryManager().synchronizeAllEpss(epssList);
-    }
-
     public Epss getEpssByCveId(String cveId) {
         return getEpssQueryManager().getEpssByCveId(cveId);
     }

apiserver/src/main/java/org/dependencytrack/persistence/EpssQueryManager.java

@@ -0,0 +1,63 @@
+/*
+ * 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.persistence.jdbi;
+
+import org.dependencytrack.model.Epss;
+import org.jdbi.v3.core.statement.Update;
+import org.jdbi.v3.sqlobject.SqlObject;
+
+import java.math.BigDecimal;
+import java.util.ArrayList;
+import java.util.Collection;
+
+/**
+ * @since 5.7.0
+ */
+public interface EpssDao extends SqlObject {
+
+    default int createOrUpdateAll(final Collection<Epss> epssRecords) {
+        final Update update = getHandle().createUpdate("""
+                INSERT INTO "EPSS" ("CVE", "SCORE", "PERCENTILE")
+                SELECT * FROM UNNEST(:cves, :scores, :percentiles)
+                ON CONFLICT ("CVE") DO UPDATE
+                SET "SCORE" = EXCLUDED."SCORE"
+                  , "PERCENTILE" = EXCLUDED."PERCENTILE"
+                WHERE "EPSS"."SCORE" IS DISTINCT FROM EXCLUDED."SCORE"
+                   OR "EPSS"."PERCENTILE" IS DISTINCT FROM EXCLUDED."PERCENTILE"
+                """);
+
+        final var cves = new ArrayList<String>(epssRecords.size());
+        final var scores = new ArrayList<BigDecimal>(epssRecords.size());
+        final var percentiles = new ArrayList<BigDecimal>(epssRecords.size());
+
+        for (final Epss epssRecord : epssRecords) {
+            cves.add(epssRecord.getCve());
+            scores.add(epssRecord.getScore());
+            percentiles.add(epssRecord.getPercentile());
+        }
+
+        return update
+                .registerArrayType(BigDecimal.class, "numeric")
+                .bindArray("cves", String.class, cves)
+                .bindArray("scores", BigDecimal.class, scores)
+                .bindArray("percentiles", BigDecimal.class, percentiles)
+                .execute();
+    }
+
+}