SP-695 Adds threads on local cryptography scanning

Author: agustingroh

Diff for: CHANGELOG.md

@@ -2,6 +2,10 @@
 
 All notable changes to this project will be documented in this file. See [standard-version](https://github.com/conventional-changelog/standard-version) for commit guidelines.
 
+### [0.13.0](https://github.com/scanoss/scanoss.js/compare/v0.12.2...v0.13.0) (2024-05-13)
+
+### [0.12.2](https://github.com/scanoss/scanoss.js/compare/v0.12.0...v0.12.2) (2024-05-10)
+
 ### [0.12.0](https://github.com/scanoss/scanoss.js/compare/v0.11.5...v0.12.0) (2024-05-06)
 
 ### [0.11.5](https://github.com/scanoss/scanoss.js/compare/v0.11.4...v0.11.5) (2024-04-19)

Diff for: package.json

@@ -1,6 +1,6 @@
 {
   "name": "scanoss",
-  "version": "0.12.2",
+  "version": "0.13.0",
   "description": "The SCANOSS JS package provides a simple, easy to consume module for interacting with SCANOSS APIs/Engine.",
   "main": "build/main/index.js",
   "typings": "build/main/index.d.ts",

Diff for: src/cli/bin/cli-bin.ts

@@ -90,6 +90,7 @@ async function main() {
   // Options
   cryptography.addOption(new Option("-r, --rules <rules>", "Crypto rules"));
   cryptography.addOption(new Option("-o, --output <filename>", "Output result file name (optional - default stdout)"));
+  cryptography.addOption(new Option("-T, --threads <threads>", "Number of threads to use while scanning (optional - default 5)"));
 
   cryptography.action((source, options) => {
     cryptoHandler(source, options).catch((e) => {

Diff for: src/cli/commands/crypto.ts

@@ -11,17 +11,18 @@ import { DependencyFilter } from '../../sdk/tree/Filters/DependencyFilter';
 import { CryptoCfg } from '../../sdk/Cryptography/CryptoCfg';
 import fs from 'fs';
 
-export async function cryptoHandler(rootPath: string, options: any){
+export async function cryptoHandler(rootPath: string, options: any): Promise<void> {
   rootPath = rootPath.replace(/\/$/, '');  // Remove trailing slash if exists
   rootPath = rootPath.replace(/^\./, process.env.PWD);  // Convert relative path to absolute path.
   const pathIsFolder = await isFolder(rootPath);
 
   let cryptoRules = null;
   if(options.rules) cryptoRules = options.rules;
 
+  let threads = null;
+  if(options.threads) threads = options.threads;
 
-  const cryptoScanner = new CryptographyScanner(new CryptoCfg(cryptoRules));
-
+  const cryptoScanner = new CryptographyScanner(new CryptoCfg({threads, rulesPath: cryptoRules}));
 
   let fileList: Array<string> = [];
   fileList.push(rootPath);
@@ -35,7 +36,7 @@ export async function cryptoHandler(rootPath: string, options: any){
   console.log("Searching for local cryptography...")
   const results = await cryptoScanner.scan(fileList);
 
-  if (options.output) {
+  if(options.output) {
     await fs.promises.writeFile(options.output, JSON.stringify(results, null, 2));
     console.log(`Results found in ${options.output}`);
   } else {

Diff for: src/sdk/Cryptography/CryptoCfg.ts

@@ -3,14 +3,21 @@
  */
 export class CryptoCfg {
 
+  private readonly DEFAULT_THREADS = 5;
+
    private readonly rulesPath: string;
 
+   private readonly threads: number;
+
   /**
    * Creates an instance of CryptoCfg.
-   * @param rulesPath Optional. Path to the cryptography rules file.
+   * @param {Object} cfg - Configuration object.
+   * @param {number} [cfg.threads=5] - The number of threads to use. Defaults to 5 if not provided.
+   * @param {string} [cfg.rulesPath] - Optional. Path to the cryptography rules file.
    */
-   constructor(rulesPath?: string) {
-     this.rulesPath = rulesPath;
+   constructor( cfg: { threads: number, rulesPath: string }) {
+     this.rulesPath = cfg.rulesPath;
+     this.threads = cfg.threads ? Number(cfg.threads) : this.DEFAULT_THREADS;
    }
 
   /**
@@ -21,4 +28,8 @@ export class CryptoCfg {
      return this.rulesPath;
    }
 
+   public getNumberOfThreads(){
+     return this.threads;
+   }
+
 }

Diff for: src/sdk/Cryptography/CryptoProvider/LocalCrypto.ts

@@ -4,25 +4,29 @@ import {
 } from '../CryptoDef/CryptoDef';
 import fs from 'fs';
 import { CryptoAlgorithm, CryptoAlgorithmRules } from '../CryptographyTypes';
+import { ThreadPool } from '../Worker/ThreadPool';
+
 
 /**
  * Represents a CryptoCalculator used for searching cryptographic algorithms in files.
  */
 export class LocalCrypto {
 
-  private cryptoMapper : Map<string, CryptoAlgorithm>;
+  private cryptoMapper: Map<string, CryptoAlgorithm>;
 
   private cryptoRules: Map<string, RegExp>;
 
-  private readonly MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024;
+  private threads: number;
 
   /**
    * Constructs a new LocalCrypto.
    * @param cryptoRules An array of CryptoAlgorithmRules used to create the search rules.
+   * @param threads Number of threads to be use to scan local cryptography (default = 5).
    */
-  constructor(cryptoRules: Array<CryptoAlgorithmRules>) {
+  constructor(cryptoRules: Array<CryptoAlgorithmRules>, threads: number) {
     this.cryptoRules = createCryptoKeywordMapper(cryptoRules);
-    this.cryptoMapper = getCryptoMapper(cryptoRules)
+    this.cryptoMapper = getCryptoMapper(cryptoRules);
+    this.threads = threads;
   }
 
   /**
@@ -31,47 +35,11 @@ export class LocalCrypto {
    */
   public async search(files: Array<string>): Promise<Array<CryptoItem>> {
     if (files.length <= 0) return [];
-    const cryptoItems = files.map((f)=> { return new CryptoItem(f) });
-
-    for(let c of cryptoItems) {
-      await this.searchCrypto(c);
-    }
-
-    return cryptoItems;
-  }
-
-  /**
-   * Asynchronously searches for cryptographic algorithms in the content of a file.
-   * @param cryptoItem The CryptoItem to search for cryptographic algorithms.
-   * @returns A promise that resolves when the search is complete.
-   */
-  private async searchCrypto(cryptoItem: CryptoItem){
-    const cryptoFound = new Array<string>();
-    const stats = await fs.promises.stat(cryptoItem.getPath());
-    if (stats.size > this.MAX_FILE_SIZE) {
-      cryptoItem.setAlgorithms([]);
-      return;
-    }
-    let content = await fs.promises.readFile(cryptoItem.getPath(), 'utf-8');
-    this.cryptoRules.forEach((value, key) => {
-      try {
-        const matches = content.match(value);
-        if (matches) {
-          cryptoFound.push(key);
-        }
-      } catch (e){
-        console.error(e);
-      }
-    });
-    // Release memory
-    content = null;
-    const results: Array<CryptoAlgorithm> = [];
-    cryptoFound.forEach((cf)=>{
-      results.push(this.cryptoMapper.get(cf));
+    const threadPool = new ThreadPool(this.threads, this.cryptoRules, this.cryptoMapper);
+    files.forEach((f) => {
+      threadPool.enqueueTask(new CryptoItem(f))
     });
-    cryptoItem.setAlgorithms(results);
+    await threadPool.init();
+    return await threadPool.processQueue();
   }
-
 }
-
-

Diff for: src/sdk/Cryptography/CryptographyScanner.ts

@@ -32,7 +32,7 @@ export class CryptographyScanner {
    */
   public async scan(files: Array<string>): Promise<ILocalCryptographyResponse> {
     const cryptographyRules = await this.loadRules(this.cryptoConfig.getRulesPath());
-    const localCrypto = new LocalCrypto(cryptographyRules);
+    const localCrypto = new LocalCrypto(cryptographyRules, this.cryptoConfig.getNumberOfThreads());
     const cryptoItems = await localCrypto.search(files);
     return mapToILocalCryptographyResponse(cryptoItems);
   }

Diff for: src/sdk/Cryptography/Worker/ThreadPool.ts

@@ -0,0 +1,110 @@
+import { CryptoAlgorithm } from '../CryptographyTypes';
+import { CryptoItem } from '../Scanneable/CryptoItem';
+import { IWorkerResponse, Worker } from './Worker';
+
+export class ThreadPool {
+
+  private readonly maxWorkers: number;
+
+  private readonly workers: Array<Worker> = [];
+
+  private readonly jobsQueue: any;
+
+  private readonly cryptoRules: Map<string, RegExp>;
+
+  private readonly cryptoMapper: Map<string, CryptoAlgorithm>;
+
+  private results = [];
+
+  private activeWorkers = 0;
+
+  private resolve;
+
+  private reject;
+  constructor(maxWorkers = 3, rules:Map<string, RegExp>, cryptoMapper: Map<string, CryptoAlgorithm> ) {
+    this.maxWorkers = maxWorkers;
+    this.workers = [];
+    this.jobsQueue = [];
+    this.cryptoRules = rules;
+    this.cryptoMapper = cryptoMapper;
+  }
+
+  enqueueTask(item: CryptoItem) {
+    return new Promise((resolve, reject) => {
+      const job = { item,  resolve, reject };
+      this.jobsQueue.push(job);
+    });
+  }
+
+  async init(): Promise<void> {
+    for (let i = 0; i < this.maxWorkers; i++) {
+      const worker = new Worker();
+      worker.on('message', async(item: IWorkerResponse) => {
+        this.results.push(item.result);
+        this.releaseWorker(item.id);
+        await this.next();
+      });
+
+      // TODO: See what can be done in case an error on the thread
+      worker.on('error', async (error) => {
+        console.log(error);
+        this.releaseWorker(worker.getId());
+        await this.next();
+      });
+
+      this.workers.push(worker);
+    }
+  }
+
+  private releaseWorker(id: number): number {
+    const wId = this.workers.findIndex(w => w.getId() === id);
+    const w = this.workers[wId];
+    w.release();
+    return w.getId();
+  }
+
+  private async next(){
+    this.activeWorkers -= 1;
+    if (this.activeWorkers === 0 && this.jobsQueue.length === 0) {
+      await this.destroyAllWorkers();
+      this.resolve(this.results);
+    } else {
+      this.processItem();
+    }
+  }
+
+  private async destroyAllWorkers() {
+    for (const worker of this.workers) {
+      await worker.terminate(); // Terminate each worker
+    }
+  }
+
+  async processQueue(): Promise<Array<CryptoItem>> {
+    return new Promise(async(resolve, reject) => {
+      this.resolve = resolve;
+      this.reject = reject;
+      this.processItem();
+    });
+  }
+
+  private processItem() {
+    if (this.workers.length > 0 && this.jobsQueue.length > 0) {
+      const freeWorkerIndices = this.workers.reduce((indices, worker, index) => {
+        if (worker.isFree()) {
+          indices.push(index);
+        }
+        return indices;
+      }, []);
+
+      freeWorkerIndices.forEach(workerIndex => {
+        if (this.jobsQueue.length <= 0) return;
+        const { item, reject } = this.jobsQueue.shift();
+        const worker = this.workers[workerIndex];
+        worker.run({ item, rules: this.cryptoRules, cryptoMapper: this.cryptoMapper });
+        this.activeWorkers += 1;
+      });
+    }
+  }
+}
+
+

Diff for: src/sdk/Cryptography/Worker/Worker.ts

@@ -0,0 +1,83 @@
+import { TransferListItem, Worker as W, WorkerOptions } from 'worker_threads';
+import { URL } from 'node:url';
+import { CryptoItem } from '../Scanneable/CryptoItem';
+
+const stringWorker = `
+const { parentPort } = require('worker_threads');
+const fs = require('fs');
+
+parentPort.on('message', async (data) => {
+
+    const MAX_FILE_SIZE = 2 * 1024 * 1024 * 1024;
+
+    const { item, rules , cryptoMapper, id } = data;
+
+     const cryptoFound = new Array();
+     const stats = await fs.promises.stat(item.file);
+     if (stats.size > MAX_FILE_SIZE) {
+      item.algorithms = [];
+      parentPort.postMessage({ result: item, id });
+      return;
+     }
+
+      let content =  fs.readFileSync(item.file, 'utf-8');
+       rules.forEach((value, key) => {
+      try {
+        const matches = content.match(value);
+        if (matches) {
+          cryptoFound.push(key);
+        }
+      } catch (e){
+        console.error(e);
+      }
+    });
+    const results = [];
+    cryptoFound.forEach((cf)=>{
+      results.push(cryptoMapper.get(cf));
+    });
+    item.algorithms = results;
+    parentPort.postMessage({ result: item, id });
+  });
+`;
+
+export interface IWorkerResponse {
+  result: CryptoItem;
+  id: number;
+}
+
+export class Worker extends W {
+
+  private free: boolean ;
+
+
+  constructor() {
+    super(stringWorker, { eval: true });
+    this.free = true;
+  }
+
+  public getId(): number {
+    return this.threadId;
+  }
+
+  public release(){
+    this.free = true;
+  }
+
+  public isFree(): boolean {
+    return this.free;
+  }
+
+  on(event, listener) {
+    if (event === 'error') {
+      this.free = true;
+    }
+    // Call super.on with the provided arguments
+     return super.on(event, listener);
+
+  }
+  public run (value: any, transferList?: ReadonlyArray<TransferListItem>){
+    this.free = false;
+    this.postMessage({...value, id: this.threadId });
+  }
+
+}

Diff for: src/sdk/Cryptography/utils/adapters/cryptoAdapters.ts

@@ -7,7 +7,7 @@ import { ICryptoItem, ILocalCryptographyResponse } from '../../CryptographyTypes
  * @returns An ILocalCryptographyResponse object containing mapped cryptographic items.
  */
 export function mapToILocalCryptographyResponse(ci: Array<CryptoItem>): ILocalCryptographyResponse {
-  const fileList: Array<ICryptoItem> = ci.map((c)=> ({  file: c.getPath(),  algorithms: c.getAlgorithms() }));
+  const fileList: Array<ICryptoItem> = ci.map((c)=> ({  file: c.file,  algorithms: c.algorithms }));
   return {
     fileList
   }