Migrate vul chain finder to data processing platform

.gitignore

@@ -6,5 +6,7 @@
 **/*.iml
 **/*.swp
 **/*.versionsBackup
+**/*.ipr
+**/*.iws
 
 

infrastructure/infrastructure/src/main/java/eu/f4sten/infra/kafka/DefaultTopics.java

@@ -24,5 +24,5 @@ private DefaultTopics() {
     public static final String INGEST = "fasten.mvn.releases";
     public static final String POM_ANALYZER = "fasten.POMAnalyzer";
     public static final String CALLABLE_INDEXER = "fasten.CallableIndexFastenPlugin";
-
+    public static final String VUL_CHAIN_FINDER = "fasten.VulChainFinder";
 }

infrastructure/infrastructure/src/test/java/eu/f4sten/infra/kafka/DefaultTopicsTest.java

@@ -25,5 +25,7 @@ public class DefaultTopicsTest {
     public void defaultValues() {
         assertEquals("fasten.mvn.releases", DefaultTopics.INGEST);
         assertEquals("fasten.POMAnalyzer", DefaultTopics.POM_ANALYZER);
+        assertEquals("fasten.CallableIndexFastenPlugin", DefaultTopics.CALLABLE_INDEXER);
+        assertEquals("fasten.VulChainFinder", DefaultTopics.VUL_CHAIN_FINDER);
     }
 }

infrastructure/loader/pom.xml

@@ -1,6 +1,5 @@
 <?xml version="1.0" encoding="UTF-8"?>
-<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
-    xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
     <modelVersion>4.0.0</modelVersion>
     <parent>
         <groupId>eu.fasten-project</groupId>
@@ -70,7 +69,11 @@
             <artifactId>ingested-artifact-completion</artifactId>
             <version>0.0.8-SNAPSHOT</version>
         </dependency>
-
+        <dependency>
+            <groupId>eu.fasten-project</groupId>
+            <artifactId>vulnerable-chain-finder</artifactId>
+            <version>0.0.8-SNAPSHOT</version>
+        </dependency>
         <!-- actual dependencies -->
         <dependency>
             <groupId>org.reflections</groupId>

plugins/pom.xml

@@ -10,6 +10,7 @@
     <packaging>pom</packaging>
 
     <modules>
+        <module>vulnerable-chain-finder</module>
         <module>examples</module>
         <module>maven-crawler</module>
         <module>pom-analyzer</module>

plugins/vulnerable-chain-finder/pom.xml

@@ -0,0 +1,32 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<project xmlns="http://maven.apache.org/POM/4.0.0"
+         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>eu.fasten-project</groupId>
+        <artifactId>plugins</artifactId>
+        <version>0.0.8-SNAPSHOT</version>
+    </parent>
+    <artifactId>vulnerable-chain-finder</artifactId>
+
+    <dependencies>
+        <dependency>
+            <groupId>eu.fasten-project</groupId>
+            <artifactId>pom-analyzer</artifactId>
+            <version>0.0.8-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>eu.fasten.analyzer</groupId>
+            <artifactId>javacg-opal</artifactId>
+            <version>0.0.8</version>
+        </dependency>
+        <dependency>
+            <groupId>eu.fasten-project</groupId>
+            <artifactId>infrastructure-impl</artifactId>
+            <version>0.0.8-SNAPSHOT</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

plugins/vulnerable-chain-finder/src/main/java/eu/f4sten/vulchainfinder/Main.java

@@ -0,0 +1,187 @@
+/*
+ * Copyright 2021 Delft University of Technology
+ *
+ * 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.
+ */
+
+package eu.f4sten.vulchainfinder;
+
+import eu.f4sten.infra.AssertArgs;
+import eu.f4sten.infra.Plugin;
+import eu.f4sten.infra.json.TRef;
+import eu.f4sten.infra.kafka.Kafka;
+import eu.f4sten.infra.kafka.Lane;
+import eu.f4sten.infra.kafka.Message;
+import eu.f4sten.infra.kafka.MessageGenerator;
+import eu.f4sten.pomanalyzer.data.MavenId;
+import eu.f4sten.vulchainfinder.exceptions.RestApiError;
+import eu.f4sten.vulchainfinder.utils.DatabaseUtils;
+import eu.f4sten.vulchainfinder.utils.ImpactPropagator;
+import eu.f4sten.vulchainfinder.utils.RestAPIDependencyResolver;
+import eu.fasten.core.data.callableindex.RocksDao;
+import eu.fasten.core.maven.data.Pom;
+import eu.fasten.core.merge.CGMerger;
+import eu.fasten.core.vulchains.VulnerableCallChain;
+import eu.fasten.core.vulchains.VulnerableCallChainRepository;
+
+import java.io.File;
+import java.util.HashSet;
+import java.util.Set;
+import javax.inject.Inject;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class Main implements Plugin {
+
+    private static final Logger LOG = LoggerFactory.getLogger(Main.class);
+    public final RestAPIDependencyResolver resolver;
+    private final DatabaseUtils db;
+    private final RocksDao dao;
+    private final Kafka kafka;
+    private final VulChainFinderArgs args;
+    private final MessageGenerator msgs;
+    private final VulnerableCallChainRepository repo;
+
+    private MavenId curId;
+
+    @Inject
+    public Main(DatabaseUtils db, RocksDao dao, Kafka kafka, VulChainFinderArgs args, MessageGenerator msgs,
+            RestAPIDependencyResolver resolver, VulnerableCallChainRepository repo) {
+        this.db = db;
+        this.dao = dao;
+        this.kafka = kafka;
+        this.args = args;
+        this.msgs = msgs;
+        this.resolver = resolver;
+        this.repo = repo;
+    }
+
+    @Override
+    public void run() {
+        AssertArgs.assertFor(args) //
+                .notNull(a -> a.kafkaIn, "kafka input topic") //
+                .notNull(a -> a.kafkaOut, "kafka output topic");
+
+        LOG.info("Subscribing to '{}', will publish in '{}' ...", args.kafkaIn, args.kafkaOut);
+
+        final var msgClass = new TRef<Message<Message<Message<Message<MavenId, Pom>, Object>, Object>, Object>>() {};
+
+        kafka.subscribe(args.kafkaIn, msgClass, (msg, l) -> {
+            final var pom = msg.input.input.input.payload;
+            curId = extractMavenIdFrom(pom);
+            LOG.info("Consuming next record ...");
+            runOrPublishErr(this::process);
+        });
+        while (true) {
+            LOG.debug("Polling ...");
+            kafka.poll();
+        }
+    }
+
+    public void process() {
+        // NOTE: this can be a temporary FS-based check and can be replaced with a better approach or removed at all.
+        if (isCurIdProcessed()) {
+            LOG.info("Coordinate {} already processed!", curId.asCoordinate());
+            return;
+        }
+
+        LOG.info("Processing {}", curId.asCoordinate());
+
+        final var allDeps = resolver.resolveDependencyIds(curId);
+
+        final var vulDeps = db.selectVulnerablePackagesExistingIn(allDeps);
+
+        Set<VulnerableCallChain> vulChains = new HashSet<>();
+        if (curIdIsPackageLevelVulnerable(vulDeps)) {
+            vulChains = extractVulCallChains(allDeps, vulDeps);
+        }
+
+        curIdIsMethodLevelVulnerable(vulChains);
+        // NOTE: it stores empty vuln. chains too to avoid re-processing records.
+        storeInVulRepo(vulChains);
+    }
+
+    private boolean curIdIsMethodLevelVulnerable(final Set<VulnerableCallChain> vulChains) {
+        return !vulChains.isEmpty();
+    }
+
+    private boolean curIdIsPackageLevelVulnerable(final Set<Long> vulDeps) {
+        return vulDeps != null && !vulDeps.isEmpty();
+    }
+
+    private void storeInVulRepo(final Set<VulnerableCallChain> vulnerableCallChains) {
+        final var productName = String.format("%s:%s", curId.groupId, curId.artifactId);
+        repo.store(productName, curId.version, vulnerableCallChains);
+    }
+
+    private Set<VulnerableCallChain> extractVulCallChains(final Set<Long> allDeps, final Set<Long> vulDeps) {
+        Set<VulnerableCallChain> result = new HashSet<>();
+
+        final var merger = new CGMerger(allDeps, db.getContext(), dao);
+        final var mergedCG = merger.mergeAllDeps();
+        final var vulCallables = db.selectVulCallablesOf(vulDeps);
+        final var propagator = new ImpactPropagator(mergedCG, merger.getAllUrisFromDB(mergedCG));
+        propagator.propagateUrisImpacts(vulCallables.keySet());
+        LOG.info("Found {} distinct vulnerable paths", propagator.getImpacts().size());
+
+        if (!propagator.getImpacts().isEmpty()) {
+            result = propagator.extractApplicationVulChains(vulCallables, curId);
+        }
+
+        return result;
+    }
+
+    public static MavenId extractMavenIdFrom(final Pom pom) {
+        final var id = new MavenId();
+        id.groupId = pom.groupId;
+        id.artifactId = pom.artifactId;
+        id.version = pom.version;
+        return id;
+    }
+
+    private void runOrPublishErr(final Runnable r) {
+        try {
+            r.run();
+        } catch (RestApiError e) {
+            LOG.error("Forced to stop the plug-in as the REST API is unavailable", e);
+            throw e;
+        } catch (Exception e) {
+            LOG.warn("Execution failed for input: {}", curId, e);
+
+            var msg = msgs.getErr(curId, returnCause(e));
+            kafka.publish(msg, args.kafkaOut, Lane.ERROR);
+        }
+    }
+
+    private Throwable returnCause(final Exception e) {
+        final var isRunTime = RuntimeException.class.equals(e.getClass());
+        final var causeNotNull = e.getCause() != null;
+        if (isRunTime && causeNotNull) {
+            return e.getCause();
+        }
+        return e;
+    }
+
+    public MavenId getCurId() {
+        return curId;
+    }
+
+    public void setCurId(final MavenId curId) {
+        this.curId = curId;
+    }
+
+    private boolean isCurIdProcessed() {
+        return new File(repo.getFilePath(String.format("%s:%s", curId.groupId, curId.artifactId), curId.version)).exists();
+    }
+
+}

plugins/vulnerable-chain-finder/src/main/java/eu/f4sten/vulchainfinder/VulChainFinderArgs.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright 2021 Delft University of Technology
+ *
+ * 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.
+ */
+package eu.f4sten.vulchainfinder;
+
+import com.beust.jcommander.Parameter;
+import eu.f4sten.infra.kafka.DefaultTopics;
+
+import java.io.File;
+
+public class VulChainFinderArgs {
+
+    @Parameter(names = "--vulchainfinder.kafkaIn", arity = 1)
+    public String kafkaIn = DefaultTopics.CALLABLE_INDEXER;
+
+    @Parameter(names = "--vulchainfinder.kafkaOut", arity = 1)
+    public String kafkaOut = DefaultTopics.VUL_CHAIN_FINDER;
+
+    @Parameter(names = "--restApiBaseUrl", arity = 1)
+    public String restApiBaseURL;
+
+    @Parameter(names = "--callableIndexPath", arity = 1)
+    public File callableIndexPath;
+
+    @Parameter(names = "--vulnChainRepoPath", arity = 1)
+    public File vulnChainRepoPath;
+}