Feature: Integrity check for maven, npm and pypi

src/main/java/org/dependencytrack/event/ComponentIntegrityCheckEvent.java

@@ -0,0 +1,17 @@
+package org.dependencytrack.event;
+
+import alpine.event.framework.Event;
+import com.github.packageurl.PackageURL;
+import org.dependencytrack.model.Component;
+
+import java.util.Optional;
+import java.util.UUID;
+
+public record ComponentIntegrityCheckEvent(String purl, Boolean internal, String md5, String sha1,
+                                           String sha256, UUID uuid, long componentId, String purlCoordinates) implements Event {
+    public ComponentIntegrityCheckEvent(final Component component) {
+        this(Optional.ofNullable(component.getPurl()).map(PackageURL::canonicalize).orElse(null), component.isInternal(), component.getMd5(),
+                component.getSha1(), component.getSha256(), component.getUuid(), component.getId(), Optional.ofNullable(component.getPurlCoordinates()).map(PackageURL::canonicalize).orElse(null));
+    }
+
+}

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

@@ -1,5 +1,7 @@
 package org.dependencytrack.event.kafka;
 
+import alpine.common.logging.Logger;
+import org.dependencytrack.event.ComponentIntegrityCheckEvent;
 import org.dependencytrack.event.ComponentRepositoryMetaAnalysisEvent;
 import org.dependencytrack.event.ComponentVulnerabilityAnalysisEvent;
 import org.dependencytrack.event.kafka.KafkaTopics.Topic;
@@ -19,6 +21,8 @@
  */
 final class KafkaEventConverter {
 
+    private static final Logger LOGGER = Logger.getLogger(KafkaEventDispatcher.class);
+
     private KafkaEventConverter() {
     }
 
@@ -55,14 +59,36 @@ static KafkaEvent<String, AnalysisCommand> convert(final ComponentRepositoryMeta
         final var componentBuilder = org.hyades.proto.repometaanalysis.v1.Component.newBuilder()
                 .setPurl(event.purlCoordinates());
         Optional.ofNullable(event.internal()).ifPresent(componentBuilder::setInternal);
-
         final var analysisCommand = AnalysisCommand.newBuilder()
                 .setComponent(componentBuilder)
                 .build();
-
+        LOGGER.debug("Dispatching repo meta analysis event for component:" + componentBuilder.getUuid());
         return new KafkaEvent<>(KafkaTopics.REPO_META_ANALYSIS_COMMAND, event.purlCoordinates(), analysisCommand, null);
     }
 
+    static KafkaEvent<String, AnalysisCommand> convert(final ComponentIntegrityCheckEvent event) {
+        if (event == null || event.purl() == null) {
+            return null;
+        }
+
+        final var componentBuilder = org.hyades.proto.repometaanalysis.v1.Component.newBuilder()
+                .setPurl(event.purl());
+        Optional.ofNullable(event.internal()).ifPresent(componentBuilder::setInternal);
+        String uuid = null;
+        if (event.uuid() != null) {
+            uuid = event.uuid().toString();
+        }
+        Optional.ofNullable(uuid).ifPresent(componentBuilder::setUuid);
+        Optional.ofNullable(event.md5()).ifPresent(componentBuilder::setMd5Hash);
+        Optional.ofNullable(event.sha1()).ifPresent(componentBuilder::setSha1Hash);
+        Optional.ofNullable(event.sha256()).ifPresent(componentBuilder::setSha256Hash);
+        final var analysisCommand = AnalysisCommand.newBuilder()
+                .setComponent(componentBuilder)
+                .build();
+        LOGGER.debug("Dispatching integrity check event for component:" + componentBuilder.getUuid());
+        return new KafkaEvent<>(KafkaTopics.INTEGRITY_ANALYSIS_COMMAND, event.purlCoordinates(), analysisCommand, null);
+    }
+
     static KafkaEvent<String, Notification> convert(final UUID projectUuid, final alpine.notification.Notification alpineNotification) {
         final Notification notification = NotificationModelConverter.convert(alpineNotification);
 

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

@@ -10,6 +10,7 @@
 import org.apache.kafka.common.KafkaException;
 import org.apache.kafka.common.errors.SerializationException;
 import org.apache.kafka.common.serialization.Serde;
+import org.dependencytrack.event.ComponentIntegrityCheckEvent;
 import org.dependencytrack.event.ComponentRepositoryMetaAnalysisEvent;
 import org.dependencytrack.event.ComponentVulnerabilityAnalysisEvent;
 import org.dependencytrack.event.GitHubAdvisoryMirrorEvent;
@@ -69,6 +70,9 @@ public Future<RecordMetadata> dispatchAsync(final Event event, final Callback ca
         } else if (event instanceof final ComponentRepositoryMetaAnalysisEvent e) {
             LOGGER.debug("Dispatch internal called for component: " + e.purlCoordinates() + " Component is internal: " + e.internal());
             return dispatchAsyncInternal(KafkaEventConverter.convert(e), callback);
+        } else if (event instanceof final ComponentIntegrityCheckEvent e) {
+            LOGGER.debug("Dispatching integrity check event for : " + e.purl() + " Component id is: " + e.componentId());
+            return dispatchAsyncInternal(KafkaEventConverter.convert(e), callback);
         } else if (event instanceof final OsvMirrorEvent e) {
             return dispatchAsyncInternal(new KafkaEvent<>(KafkaTopics.VULNERABILITY_MIRROR_COMMAND, Vulnerability.Source.OSV.name(), e.ecosystem(), null), callback);
         } else if (event instanceof NistMirrorEvent) {

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

@@ -22,6 +22,7 @@
 import org.dependencytrack.event.ProjectMetricsUpdateEvent;
 import org.dependencytrack.event.ProjectPolicyEvaluationEvent;
 import org.dependencytrack.event.kafka.processor.DelayedBomProcessedNotificationProcessor;
+import org.dependencytrack.event.kafka.processor.IntegrityAnalysisResultProcessor;
 import org.dependencytrack.event.kafka.processor.MirrorVulnerabilityProcessor;
 import org.dependencytrack.event.kafka.processor.RepositoryMetaResultProcessor;
 import org.dependencytrack.event.kafka.processor.VulnerabilityScanResultProcessor;
@@ -235,6 +236,12 @@ Topology createTopology() {
                                 .withName("consume_from_%s_topic".formatted(KafkaTopics.REPO_META_ANALYSIS_RESULT.name())))
                 .process(RepositoryMetaResultProcessor::new, Named.as("process_repo_meta_analysis_result"));
 
+        streamsBuilder
+                .stream(KafkaTopics.INTEGRITY_ANALYSIS_RESULT.name(),
+                        Consumed.with(KafkaTopics.INTEGRITY_ANALYSIS_RESULT.keySerde(), KafkaTopics.INTEGRITY_ANALYSIS_RESULT.valueSerde())
+                                .withName("consume_from_%s_topic".formatted(KafkaTopics.INTEGRITY_ANALYSIS_RESULT.name())))
+                .process(IntegrityAnalysisResultProcessor::new, Named.as("process_component_integrity_analysis_result"));
+
         streamsBuilder
                 .stream(KafkaTopics.NEW_VULNERABILITY.name(),
                         Consumed.with(KafkaTopics.NEW_VULNERABILITY.keySerde(), KafkaTopics.NEW_VULNERABILITY.valueSerde())

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

@@ -7,6 +7,7 @@
 import org.dependencytrack.common.ConfigKey;
 import org.dependencytrack.event.kafka.serialization.KafkaProtobufSerde;
 import org.hyades.proto.notification.v1.Notification;
+import org.hyades.proto.repointegrityanalysis.v1.IntegrityResult;
 import org.hyades.proto.repometaanalysis.v1.AnalysisCommand;
 import org.hyades.proto.repometaanalysis.v1.AnalysisResult;
 import org.hyades.proto.vulnanalysis.v1.ScanCommand;
@@ -31,10 +32,13 @@ public final class KafkaTopics {
     public static final Topic<String, String> VULNERABILITY_MIRROR_COMMAND;
     public static final Topic<String, Bom> NEW_VULNERABILITY;
     public static final Topic<String, AnalysisCommand> REPO_META_ANALYSIS_COMMAND;
+
+    public static final Topic<String, AnalysisCommand> INTEGRITY_ANALYSIS_COMMAND;
     public static final Topic<String, AnalysisResult> REPO_META_ANALYSIS_RESULT;
     public static final Topic<ScanKey, ScanCommand> VULN_ANALYSIS_COMMAND;
     public static final Topic<ScanKey, ScanResult> VULN_ANALYSIS_RESULT;
 
+    public static final Topic<String, IntegrityResult>INTEGRITY_ANALYSIS_RESULT;
     public static final Topic<String, Notification> NOTIFICATION_PROJECT_VULN_ANALYSIS_COMPLETE;
     private static final Serde<Notification> NOTIFICATION_SERDE = new KafkaProtobufSerde<>(Notification.parser());
 
@@ -55,6 +59,8 @@ public final class KafkaTopics {
         VULNERABILITY_MIRROR_COMMAND = new Topic<>("dtrack.vulnerability.mirror.command", Serdes.String(), Serdes.String());
         NEW_VULNERABILITY = new Topic<>("dtrack.vulnerability", Serdes.String(), new KafkaProtobufSerde<>(Bom.parser()));
         REPO_META_ANALYSIS_COMMAND = new Topic<>("dtrack.repo-meta-analysis.component", Serdes.String(), new KafkaProtobufSerde<>(AnalysisCommand.parser()));
+        INTEGRITY_ANALYSIS_COMMAND = new Topic<>("dtrack.integrity-analysis.component", Serdes.String(), new KafkaProtobufSerde<>(AnalysisCommand.parser()));
+        INTEGRITY_ANALYSIS_RESULT = new Topic<>("dtrack.integrity-analysis.result", Serdes.String(), new KafkaProtobufSerde<>(IntegrityResult.parser()));
         REPO_META_ANALYSIS_RESULT = new Topic<>("dtrack.repo-meta-analysis.result", Serdes.String(), new KafkaProtobufSerde<>(AnalysisResult.parser()));
         VULN_ANALYSIS_COMMAND = new Topic<>("dtrack.vuln-analysis.component", new KafkaProtobufSerde<>(ScanKey.parser()), new KafkaProtobufSerde<>(ScanCommand.parser()));
         VULN_ANALYSIS_RESULT = new Topic<>("dtrack.vuln-analysis.result", new KafkaProtobufSerde<>(ScanKey.parser()), new KafkaProtobufSerde<>(ScanResult.parser()));

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

@@ -0,0 +1,116 @@
+package org.dependencytrack.event.kafka.processor;
+
+import alpine.common.logging.Logger;
+import alpine.common.metrics.Metrics;
+import com.github.packageurl.MalformedPackageURLException;
+import com.github.packageurl.PackageURL;
+import io.micrometer.core.instrument.Timer;
+import org.apache.kafka.streams.processor.api.Processor;
+import org.apache.kafka.streams.processor.api.Record;
+import org.dependencytrack.model.Component;
+import org.dependencytrack.model.ComponentIntegrityAnalysis;
+import org.dependencytrack.persistence.QueryManager;
+import org.hyades.proto.repointegrityanalysis.v1.HashMatchStatus;
+import org.hyades.proto.repointegrityanalysis.v1.IntegrityResult;
+
+import javax.jdo.JDODataStoreException;
+import javax.jdo.PersistenceManager;
+import javax.jdo.Query;
+import javax.jdo.Transaction;
+import java.util.Date;
+import java.util.UUID;
+
+public class IntegrityAnalysisResultProcessor implements Processor<String, IntegrityResult, Void, Void> {
+    private static final Logger LOGGER = Logger.getLogger(IntegrityAnalysisResultProcessor.class);
+    private static final Timer TIMER = Timer.builder("integrity_analysis_result_processing")
+            .description("Time taken to process integrity analysis results")
+            .register(Metrics.getRegistry());
+
+    @Override
+    public void process(Record<String, IntegrityResult> record) {
+        final Timer.Sample timerSample = Timer.start();
+        try (final var qm = new QueryManager()) {
+            synchronizeComponentIntegrity(qm.getPersistenceManager(), record);
+        } catch (Exception e) {
+            LOGGER.error("An unexpected error occurred while processing record %s".formatted(record), e);
+        } finally {
+            timerSample.stop(TIMER);
+        }
+    }
+
+    private void synchronizeComponentIntegrity(final PersistenceManager pm, final Record<String, IntegrityResult> record) {
+
+        final IntegrityResult result = record.value();
+
+        final PackageURL purl;
+        try {
+            purl = new PackageURL(result.getComponent().getPurl());
+        } catch (MalformedPackageURLException e) {
+            LOGGER.warn("""
+                    Received repository integrity information with invalid PURL,\s
+                    will not be able to correlate; Dropping
+                    """, e);
+            return;
+        }
+
+        final Transaction trx = pm.currentTransaction();
+        try {
+            trx.begin();
+            final Query<ComponentIntegrityAnalysis> query = pm.newQuery(ComponentIntegrityAnalysis.class);
+            query.setFilter("repositoryIdentifier == :repository && component.uuid == :uuid");
+            query.setParameters(
+                    record.value().getRepositoryIdentifier(),
+                    UUID.fromString(record.value().getComponent().getUuid())
+            );
+            ComponentIntegrityAnalysis persistentIntegrityResult = query.executeUnique();
+            if (persistentIntegrityResult == null || persistentIntegrityResult.getComponent() == null) {
+                persistentIntegrityResult = new ComponentIntegrityAnalysis();
+            }
+
+            if (persistentIntegrityResult.getLastCheck() != null
+                    && persistentIntegrityResult.getLastCheck().after(new Date(record.timestamp()))) {
+                LOGGER.warn("""
+                        Received integrity check information for %s that is older\s
+                        than what's already in the database; Discarding
+                        """.formatted(purl));
+                return;
+            }
+            final Query<Component> queryComponent = pm.newQuery(Component.class);
+            queryComponent.setFilter("uuid == :uuid");
+            queryComponent.setParameters(UUID.fromString(record.value().getComponent().getUuid()));
+            Component component = queryComponent.executeUnique();
+            if (component != null) {
+                persistentIntegrityResult.setRepositoryIdentifier(record.value().getRepositoryIdentifier());
+                HashMatchStatus md5HashMatch = record.value().getMd5HashMatch();
+                HashMatchStatus sha1HashMatch = record.value().getSha1HashMatch();
+                HashMatchStatus sha256HashMatch = record.value().getSha256HashMatch();
+                persistentIntegrityResult.setMd5HashMatched(md5HashMatch.name());
+                persistentIntegrityResult.setSha256HashMatched(sha256HashMatch.name());
+                persistentIntegrityResult.setSha1HashMatched(sha1HashMatch.name());
+                persistentIntegrityResult.setComponent(component);
+                persistentIntegrityResult.setLastCheck(new Date(record.timestamp()));
+                if (md5HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_FAIL) || sha1HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_FAIL) || sha256HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_FAIL)) {
+                    persistentIntegrityResult.setIntegrityCheckPassed(false);
+                } else if (md5HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_UNKNOWN) && sha1HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_UNKNOWN) && sha256HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_UNKNOWN)) {
+                    persistentIntegrityResult.setIntegrityCheckPassed(false);
+                } else if (md5HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_COMPONENT_MISSING_HASH) && sha1HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_COMPONENT_MISSING_HASH) && sha256HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_COMPONENT_MISSING_HASH)) {
+                    persistentIntegrityResult.setIntegrityCheckPassed(false);
+                } else {
+                    boolean flag = (md5HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_PASS) || md5HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_UNKNOWN))
+                            && (sha1HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_PASS) || sha1HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_UNKNOWN))
+                            && (sha256HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_PASS) || sha256HashMatch.equals(HashMatchStatus.HASH_MATCH_STATUS_UNKNOWN));
+                    persistentIntegrityResult.setIntegrityCheckPassed(flag);
+                }
+                pm.makePersistent(persistentIntegrityResult);
+
+                trx.commit();
+            }
+        } catch (JDODataStoreException e) {
+            LOGGER.error("An unexpected error occurred while executing JDO query %s".formatted(record), e);
+        } finally {
+            if (trx.isActive()) {
+                trx.rollback();
+            }
+        }
+    }
+}

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

@@ -130,4 +130,5 @@ private void synchronizeRepositoryMetaComponent(final PersistenceManager pm, fin
         }
     }
 
+
 }

src/main/java/org/dependencytrack/model/Component.java

@@ -355,6 +355,7 @@ public enum FetchGroup {
     private transient String licenseId;
     private transient DependencyMetrics metrics;
     private transient RepositoryMetaComponent repositoryMeta;
+    private transient ComponentIntegrityAnalysis integrityAnalysis;
     private transient boolean isNew;
     private transient int usedBy;
     private transient Set<String> dependencyGraph;
@@ -741,6 +742,14 @@ public void setRepositoryMeta(RepositoryMetaComponent repositoryMeta) {
         this.repositoryMeta = repositoryMeta;
     }
 
+    public ComponentIntegrityAnalysis getIntegrityAnalysis() {
+        return integrityAnalysis;
+    }
+
+    public void setIntegrityAnalysis(ComponentIntegrityAnalysis integrityAnalysis) {
+        this.integrityAnalysis = integrityAnalysis;
+    }
+
     public boolean isNew() {
         return isNew;
     }