Job Streaming for PNE/Patchfinder

patchfinder/env.list

@@ -10,17 +10,18 @@ RABBIT_HOST=host.docker.internal
 RABBIT_PORT=5672
 RABBIT_USERNAME=guest
 RABBIT_PASSWORD=guest
-PF_INPUT_QUEUE=PNE_OUT
+PF_INPUT_QUEUE=PNE_OUT_PATCH
+FF_INPUT_QUEUE=PNE_OUT_FIX
 
 # ---       PATCH FINDER VARS     ---
-PF_INPUT_MODE=off
+PF_INPUT_MODE=rabbit
 CVE_LIMIT=20
 ADDRESS_BASES=https://www.github.com/,https://www.gitlab.com/
 MAX_THREADS=10
-CLONE_COMMIT_THRESHOLD=1000
-CLONE_COMMIT_LIMIT=50000
+CLONE_COMMIT_THRESHOLD=250
+CLONE_COMMIT_LIMIT=200000
 CLONE_PATH=nvip_data/patch-repos
 PATCH_SRC_URL_PATH=nvip_data/source_dict.json
 
 # ---       FIX FINDER VARS     ---
-FF_INPUT_MODE=dev
+FF_INPUT_MODE=off

patchfinder/pom.xml

@@ -9,8 +9,6 @@
     <version>1.0</version>
 
     <properties>
-        <maven.compiler.source>1.8</maven.compiler.source>
-        <maven.compiler.target>1.8</maven.compiler.target>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
     </properties>
 
@@ -39,10 +37,20 @@
                 </executions>
             </plugin>
 
+            <plugin>
+                <groupId>org.apache.maven.plugins</groupId>
+                <artifactId>maven-compiler-plugin</artifactId>
+                <version>3.11.0</version>
+                <configuration>
+                    <source>17</source>
+                    <target>17</target>
+                </configuration>
+            </plugin>
+
             <plugin>
                 <groupId>org.apache.maven.plugins</groupId>
                 <artifactId>maven-surefire-plugin</artifactId>
-                <version>2.19</version>
+                <version>2.22.2</version>
                 <!-- Other configuration options -->
             </plugin>
         </plugins>
@@ -132,6 +140,12 @@
             <scope>test</scope>
         </dependency>
 
+        <dependency>
+            <groupId>org.mockito</groupId>
+            <artifactId>mockito-junit-jupiter</artifactId>
+            <scope>test</scope>
+        </dependency>
+
         <dependency>
             <groupId>org.springframework</groupId>
             <artifactId>spring-test</artifactId>

patchfinder/src/main/java/FixFinderMain.java

@@ -22,7 +22,10 @@
  * SOFTWARE.
  */
 
+import db.DatabaseHelper;
 import env.FixFinderEnvVars;
+import env.SharedEnvVars;
+import messenger.Messenger;
 import org.apache.logging.log4j.LogManager;
 import org.apache.logging.log4j.Logger;
 import fixes.FixFinder;
@@ -37,6 +40,13 @@
  */
 public class FixFinderMain extends Thread {
     private final static Logger logger = LogManager.getLogger(FixFinderMain.class);
+    private final DatabaseHelper databaseHelper;
+    private final Messenger messenger;
+
+    public FixFinderMain(DatabaseHelper dbh, Messenger messenger) {
+        this.databaseHelper = dbh;
+        this.messenger = messenger;
+    }
 
     /**
      * Entry point for the FixFinder, initializes necessary classes and start listening for jobs with RabbitMQ
@@ -45,18 +55,24 @@ public class FixFinderMain extends Thread {
     public void run() {
         logger.info("Starting FixFinder...");
 
-        // Init FixFinder
-        FixFinder.init();
+        // Get input mode
+        final String inputMode = FixFinderEnvVars.getInputMode();
 
         // Determine run mode and start PatchFinder
-        switch (FixFinderEnvVars.getInputMode()) {
+        switch (inputMode) {
             case "db":
+                // Init FixFinder
+                FixFinder.init(this.databaseHelper);
                 runDb();
                 break;
             case "rabbit":
+                // Init FixFinder
+                FixFinder.init(this.databaseHelper);
                 runRabbit();
                 break;
             case "dev":
+                // Init FixFinder
+                FixFinder.init(this.databaseHelper);
                 runDev();
                 break;
             default:
@@ -70,32 +86,20 @@ private void runDb() {
         List<String> cveIds = new ArrayList<>(FixFinder.getDatabaseHelper().getCves(FixFinderEnvVars.getCveLimit()));
         logger.info("Successfully got {} CVEs from the database", cveIds.size());
 
-        try {
-            FixFinder.run(cveIds);
-        } catch (Exception e) {
-            logger.error("A fatal error attempting to complete jobs: {}", e.toString());
-        }
+        for (String cveId : cveIds) FixFinder.run(cveId);
     }
 
+    // TODO: Support end message
     private void runRabbit() {
-        // TODO: RabbitMQ integration, wait until PoC is accepted to complete this
-        throw new UnsupportedOperationException();
+        // Start job handling
+        messenger.startHandlingFixJobs(SharedEnvVars.getFixFinderInputQueue());
     }
 
     private void runDev() {
         // Manually enter CVEs for development
         List<String> cveIds = new ArrayList<>();
         cveIds.add("CVE-2023-38571");
 
-        try {
-            FixFinder.run(cveIds);
-        } catch (Exception e) {
-            logger.error("A fatal error attempting to complete jobs: {}", e.toString());
-        }
-    }
-
-    public static void main(String[] args) {
-        FixFinderMain finder = new FixFinderMain();
-        finder.start();
+        for (String cveId : cveIds) FixFinder.run(cveId);
     }
 }

patchfinder/src/main/java/PatchFinderMain.java

@@ -22,12 +22,13 @@
  * SOFTWARE.
  */
 
+import db.DatabaseHelper;
 import env.PatchFinderEnvVars;
+import env.SharedEnvVars;
 import messenger.Messenger;
 import model.CpeGroup;
 
 import java.io.IOException;
-import java.util.List;
 import java.util.Map;
 
 import org.apache.logging.log4j.LogManager;
@@ -41,6 +42,13 @@
  */
 public class PatchFinderMain extends Thread {
     private final static Logger logger = LogManager.getLogger(PatchFinderMain.class);
+    private final DatabaseHelper databaseHelper;
+    private final Messenger messenger;
+
+    public PatchFinderMain(DatabaseHelper dbh, Messenger messenger) {
+        this.databaseHelper = dbh;
+        this.messenger = messenger;
+    }
 
     /**
      * Entry point for the PatchFinder, initializes necessary classes and start listening for jobs with RabbitMQ
@@ -49,7 +57,7 @@ public class PatchFinderMain extends Thread {
     public void run() {
         logger.info("Starting PatchFinder...");
         // Init PatchFinder
-        PatchFinder.init();
+        PatchFinder.init(this.databaseHelper);
 
         // Determine run mode and start PatchFinder
         switch (PatchFinderEnvVars.getInputMode()) {
@@ -71,36 +79,21 @@ private void runDb() {
         final int affectedProductsCount = affectedProducts.values().stream().map(CpeGroup::getVersionsCount).reduce(0, Integer::sum);
         logger.info("Successfully got {} CVEs mapped to {} affected products from the database", affectedProducts.size(), affectedProductsCount);
         try {
-            PatchFinder.run(affectedProducts, PatchFinderEnvVars.getCveLimit());
+            // TODO: Delegate to threads
+            for (String cveId : affectedProducts.keySet()) {
+                PatchFinder.run(cveId, affectedProducts.get(cveId));
+            }
+
+            // When all threads are done, write source dict to file
+            PatchFinder.writeSourceDict();
         } catch (IOException e) {
             logger.error("A fatal error attempting to complete jobs: {}", e.toString());
         }
     }
 
+    // TODO: Support end message
     private void runRabbit() {
-        // Start busy-wait loop
-        final Messenger rabbitMQ = new Messenger(
-                PatchFinderEnvVars.getRabbitHost(),
-                PatchFinderEnvVars.getRabbitVHost(),
-                    PatchFinderEnvVars.getRabbitPort(),PatchFinderEnvVars.getRabbitUsername(),
-                PatchFinderEnvVars.getRabbitPassword(),
-                    PatchFinderEnvVars.getRabbitInputQueue()
-        );
-        logger.info("Starting busy-wait loop for jobs...");
-        while(true) {
-            try {
-                // Wait and get jobs
-                final List<String> jobs = rabbitMQ.waitForProductNameExtractorMessage(PatchFinderEnvVars.getRabbitPollInterval());
-
-                // If null is returned, either and error occurred or intentional program quit
-                if(jobs == null) break;
-
-                // Otherwise, run received jobs
-                PatchFinder.run(jobs);
-            } catch (IOException | InterruptedException e) {
-                logger.error("A fatal error occurred during job waiting: {}", e.toString());
-                break;
-            }
-        }
+        // Start job handling
+        messenger.startHandlingPatchJobs(SharedEnvVars.getPatchFinderInputQueue());
     }
 }

patchfinder/src/main/java/PatchFixMain.java

@@ -1,6 +1,30 @@
+import db.DatabaseHelper;
+import env.SharedEnvVars;
+import messenger.Messenger;
+
 public class PatchFixMain {
     public static void main(String[] args) {
-        new PatchFinderMain().start();
-        new FixFinderMain().start();
+        SharedEnvVars.initializeEnvVars(false);
+
+        // Init dbh
+        final DatabaseHelper dbh = new DatabaseHelper(
+                SharedEnvVars.getDatabaseType(),
+                SharedEnvVars.getHikariUrl(),
+                SharedEnvVars.getHikariUser(),
+                SharedEnvVars.getHikariPassword()
+        );
+
+        // Init messenger
+        final Messenger m = new Messenger(
+                SharedEnvVars.getRabbitHost(),
+                SharedEnvVars.getRabbitVHost(),
+                SharedEnvVars.getRabbitPort(),
+                SharedEnvVars.getRabbitUsername(),
+                SharedEnvVars.getRabbitPassword()
+        );
+
+        // Init and start Patchfinder/Fixfinder with dbh and messenger instances
+        new PatchFinderMain(dbh, m).start();
+        new FixFinderMain(dbh, m).start();
     }
 }

patchfinder/src/main/java/db/DatabaseHelper.java

@@ -201,10 +201,10 @@ public Set<String> getExistingPatchCommitShas() {
 	 * Collects a map of CPEs with their correlated CVE and Vuln ID used for
 	 * collecting patches given a list of CVE ids.
 	 *
-	 * @param cveIds CVEs to get affected products for
+	 * @param cveId CVEs to get affected products for
 	 * @return a map of affected products
 	 */
-	public Map<String, CpeGroup> getAffectedProducts(List<String> cveIds) {
+	public Map<String, CpeGroup> getAffectedProducts(String cveId) {
 		Map<String, CpeGroup> affectedProducts = new HashMap<>();
 		// Prepare statement
 		try (Connection conn = getConnection();
@@ -213,16 +213,14 @@ public Map<String, CpeGroup> getAffectedProducts(List<String> cveIds) {
 		) {
 			// Execute correct statement and get result set
 			ResultSet res = null;
-			if(cveIds == null) {
+			if(cveId == null) {
 				res = getAll.executeQuery();
 				parseAffectedProducts(affectedProducts, res);
 			}
 			else {
-				for (String cveId : cveIds) {
-					getById.setString(1, cveId);
-					res = getById.executeQuery();
-					parseAffectedProducts(affectedProducts, res);
-				}
+				getById.setString(1, cveId);
+				res = getById.executeQuery();
+				parseAffectedProducts(affectedProducts, res);
 			}
 
 		} catch (Exception e) {
@@ -412,6 +410,33 @@ public List<String> getCves(int cveLimit) {
 		return cves;
 	}
 
+	public int[] insertFixes(Set<Fix> fixes) {
+		int failedInserts = 0;
+		int existingInserts = 0;
+
+		for (Fix fix : fixes) {
+			try {
+				final int result = this.insertFix(fix);
+				// Result of operation, 0 for OK, 1 for failed, 2 for already exists
+				switch (result) {
+					case 1:
+						failedInserts++;
+						break;
+					case 2:
+						existingInserts++;
+						break;
+					default:
+						break;
+				}
+			}
+			catch (SQLException e) {
+				logger.error("Failed to insert fix {}: {}", fix, e.toString());
+			}
+		}
+
+		return new int[] {failedInserts, existingInserts};
+	}
+
 	/**
 	 * Method for inserting a fix into the fixes table
 	 * Should also check for duplicates