Merge pull request #62 from fosslight/correct

Correct the source/bin scanner result

Author: dd-jy

src/fosslight_scanner/_help.py

@@ -33,6 +33,8 @@
             -r\t\t\t    Keep raw data
             -t\t\t\t    Hide the progress bar
             -v\t\t\t    Print FOSSLight Scanner version
+            --no_correction\t    Enter if you don't want to correct OSS information with sbom-info.yaml
+            --correct_fpath <path>  Path to the sbom-info.yaml file
 
         Options for only 'all' or 'bin' mode
             -u <db_url>\t\t    DB Connection(format :'postgresql://username:password@host:port/database_name')

src/fosslight_scanner/cli.py

@@ -21,10 +21,14 @@ def main():
     parser.add_argument('--dependency', '-d', help='Dependency arguments', type=str, dest='dep_argument', default="")
     parser.add_argument('--url', '-u', help="DB Url", type=str, dest='db_url', default="")
     parser.add_argument('--core', '-c', help='Number of processes to analyze source', type=int, dest='core', default=-1)
-    parser.add_argument('--raw', '-r', help='Keep raw data',  action='store_true', dest='raw', default=False)
-    parser.add_argument('--timer', '-t', help='Hide the progress bar',  action='store_true', dest='timer', default=False)
-    parser.add_argument('--version', '-v', help='Print version',  action='store_true', dest='version', default=False)
+    parser.add_argument('--raw', '-r', help='Keep raw data', action='store_true', dest='raw', default=False)
+    parser.add_argument('--timer', '-t', help='Hide the progress bar', action='store_true', dest='timer', default=False)
+    parser.add_argument('--version', '-v', help='Print version', action='store_true', dest='version', default=False)
     parser.add_argument('--help', '-h', help='Print help message', action='store_true', dest='help')
+    parser.add_argument('--no_correction', help='No correction with sbom-info.yaml',
+                        action='store_true', required=False, default=False)
+    parser.add_argument('--correct_fpath', help='Path to the sbom-info.yaml',
+                        type=str, required=False, default='')
 
     try:
         args = parser.parse_args()
@@ -37,7 +41,8 @@ def main():
         print_package_version(PKG_NAME, "FOSSLight Scanner Version:")
     else:
         run_main(args.mode, args.path, args.dep_argument, args.output, args.file,
-                 args.link, args.db_url, args.timer, args.raw, args.core)
+                 args.link, args.db_url, args.timer, args.raw, args.core,
+                 not args.no_correction, args.correct_fpath)
 
 
 if __name__ == "__main__":

src/fosslight_scanner/common.py

@@ -6,21 +6,26 @@
 import os
 import sys
 import logging
-from shutil import copy
+import shutil
 import re
 import pandas as pd
 import yaml
+import copy
 import fosslight_util.constant as constant
 from fosslight_util.parsing_yaml import parsing_yml
 from fosslight_util.write_yaml import create_yaml_with_ossitem
+from fosslight_util.read_excel import read_oss_report
+from fosslight_util.output_format import write_output_file
 
 logger = logging.getLogger(constant.LOGGER_NAME)
+SRC_SHEET = 'SRC_FL_Source'
+BIN_SHEET = 'BIN_FL_Binary'
 
 
 def copy_file(source, destination):
     copied_file = ""
     try:
-        copy(source, destination)
+        shutil.copy(source, destination)
         if os.path.isdir(destination):
             copied_file = os.path.join(destination, os.path.basename(source))
         else:
@@ -194,3 +199,102 @@ def merge_yamls(_output_dir, merge_yaml_files, final_report, remove_src_data=Fal
         success = False
 
     return success, err_msg
+
+
+def correct_scanner_result(_output_dir, output_files, exist_src, exist_bin):
+    src_oss_list = []
+    bin_oss_list = []
+    duplicates = False
+
+    if exist_src:
+        src_oss_list = check_exclude_dir(get_osslist_with_xlsx(_output_dir, output_files['SRC'], SRC_SHEET))
+    if exist_bin:
+        bin_oss_list = check_exclude_dir(get_osslist_with_xlsx(_output_dir, output_files['BIN'], BIN_SHEET))
+
+    if exist_src and exist_bin:
+        try:
+            dup_bin_list = []
+            exclude_list = []
+            for src_item in src_oss_list:
+                dup_bin = ''
+                for idx, bin_item in enumerate(bin_oss_list):
+                    if not src_item.source_name_or_path:
+                        continue
+                    if src_item.source_name_or_path[0] == bin_item.source_name_or_path[0]:
+                        dup_bin = copy.deepcopy(bin_item)
+                        if not dup_bin.license:
+                            if dup_bin.exclude:
+                                src_item.exclude = dup_bin.exclude
+                            dup_bin.set_sheet_item(src_item.get_print_array()[0])
+                            if dup_bin.comment:
+                                dup_bin.comment += '/'
+                            dup_bin.comment += 'Loaded from SRC OSS info'
+                            dup_bin_list.append(dup_bin)
+                            exclude_list.append(idx)
+                if dup_bin:
+                    src_item.exclude = True
+                    if src_item.comment:
+                        src_item.comment += '/'
+                    src_item.comment += 'Excluded by duplicated binary within BIN'
+                    duplicates = True
+
+            exclude_list = list(set(exclude_list))
+            for idx in exclude_list:
+                bin_oss_list[idx].exclude = True
+                if bin_oss_list[idx].comment:
+                    bin_oss_list[idx].comment += '/'
+                bin_oss_list[idx].comment += 'Excluded by SRC OSS info'
+            bin_oss_list.extend(dup_bin_list)
+        except Exception as ex:
+            logger.warning(f"correct the scanner result:{ex}")
+
+    try:
+        if exist_src:
+            success, err_msg = write_xlsx_with_osslist(src_oss_list, _output_dir, output_files['SRC'], SRC_SHEET)
+            if not success:
+                logger.warning(err_msg)
+        if exist_bin:
+            success, err_msg = write_xlsx_with_osslist(bin_oss_list, _output_dir, output_files['BIN'], BIN_SHEET)
+            if not success:
+                logger.warning(err_msg)
+        if duplicates:
+            logger.info('Success to correct the src/bin scanner result')
+    except Exception as ex:
+        logger.warning(f"Corrected src/bin scanner result:{ex}")
+    return
+
+
+def write_xlsx_with_osslist(oss_list, output_dir, output_file, sheetname):
+    sheet_list = {}
+    sheet_list[sheetname] = []
+    new_oss_list = []
+
+    for src_item in oss_list:
+        new_oss_list.append(src_item.get_print_array()[0])
+    sheet_list[sheetname].extend(new_oss_list)
+    if os.path.exists(os.path.join(output_dir, output_file)):
+        os.remove(os.path.join(output_dir, output_file))
+    success, err_msg, result_file = write_output_file(os.path.join(output_dir, output_file).rstrip('xlsx'), '.xlsx',
+                                                      sheet_list)
+    return success, err_msg
+
+
+def get_osslist_with_xlsx(_output_dir, output_file, sheet_name):
+    oss_list = []
+    oss_xlsx = os.path.join(_output_dir, output_file)
+
+    if os.path.exists(oss_xlsx):
+        oss_list.extend(read_oss_report(oss_xlsx, sheet_name))
+
+    return oss_list
+
+
+def check_exclude_dir(oss_list):
+    _exclude_dirs = ["venv", "node_modules", "Pods", "Carthage"]
+
+    for oss_item in oss_list:
+        for exclude_dir in _exclude_dirs:
+            if exclude_dir in oss_item.source_name_or_path[0].split(os.path.sep):
+                oss_item.exclude = True
+                break
+    return oss_list

src/fosslight_scanner/fosslight_scanner.py

@@ -9,8 +9,8 @@
 import re
 import yaml
 import sys
+import shutil
 from pathlib import Path
-from shutil import rmtree as rmdir
 from datetime import datetime
 from fosslight_binary import binary_analysis
 from fosslight_dependency.run_dependency_scanner import run_dependency_scanner
@@ -23,7 +23,7 @@
 from fosslight_prechecker._precheck import run_lint as prechecker_lint
 from .common import (copy_file, call_analysis_api,
                      overwrite_excel, extract_name_from_link,
-                     merge_yamls)
+                     merge_yamls, correct_scanner_result)
 from fosslight_util.write_excel import merge_excels
 from ._run_compare import run_compare
 import subprocess
@@ -102,7 +102,8 @@ def run_scanner(src_path, dep_arguments, output_path, keep_raw_data=False,
                 run_src=True, run_bin=True, run_dep=True, run_prechecker=True,
                 remove_src_data=True, result_log={}, output_file="",
                 output_extension="", num_cores=-1, db_url="",
-                default_oss_name="", url=""):
+                default_oss_name="", url="",
+                correct_mode=True, correct_fpath=""):
     final_excel_dir = output_path
     success = True
     temp_output_fiiles = []
@@ -115,6 +116,9 @@ def run_scanner(src_path, dep_arguments, output_path, keep_raw_data=False,
     if output_extension == "":
         output_extension = ".xlsx"
 
+    if not correct_fpath:
+        correct_fpath = src_path
+
     try:
         sheet_list = {}
         final_excel_dir = os.path.abspath(final_excel_dir)
@@ -149,7 +153,8 @@ def run_scanner(src_path, dep_arguments, output_path, keep_raw_data=False,
                             sheet_list["SRC_FL_Source"] = [scan_item.get_row_to_print() for scan_item in result[2]]
                             need_license = True if output_extension == ".xlsx" else False
                             create_report_file(0, result[2], result[3], 'all', need_license,
-                                               _output_dir, output_files["SRC"].split('.')[0], output_extension)
+                                               _output_dir, output_files["SRC"].split('.')[0], output_extension,
+                                               correct_mode, correct_fpath, abs_path)
                     else:  # Run fosslight_source by using docker image
                         src_output = os.path.join("output", output_files["SRC"])
                         output_rel_path = os.path.relpath(abs_path, os.getcwd())
@@ -166,7 +171,8 @@ def run_scanner(src_path, dep_arguments, output_path, keep_raw_data=False,
                                                1, binary_analysis.find_binaries,
                                                abs_path,
                                                os.path.join(_output_dir, output_files["BIN"]),
-                                               "", db_url)
+                                               "", db_url, False,
+                                               correct_mode, correct_fpath)
                 if success:
                     output_binary_txt_raw = f"{output_files['BIN'].split('.')[0]}.txt"
                     success_file, copied_file = copy_file(os.path.join(_output_dir, output_binary_txt_raw),
@@ -186,6 +192,19 @@ def run_scanner(src_path, dep_arguments, output_path, keep_raw_data=False,
     try:
         output_file_without_ext = os.path.join(final_excel_dir, output_file)
         final_report = f"{output_file_without_ext}{output_extension}"
+        tmp_dir = f"tmp_{datetime.now().strftime('%y%m%d_%H%M')}"
+        exist_src = False
+        exist_bin = False
+        if correct_mode:
+            os.makedirs(os.path.join(_output_dir, tmp_dir), exist_ok=True)
+            if os.path.exists(os.path.join(_output_dir, output_files['SRC'])):
+                exist_src = True
+                shutil.copy2(os.path.join(_output_dir, output_files['SRC']), os.path.join(_output_dir, tmp_dir))
+            if os.path.exists(os.path.join(_output_dir, output_files['BIN'])):
+                exist_bin = True
+                shutil.copy2(os.path.join(_output_dir, output_files['BIN']), os.path.join(_output_dir, tmp_dir))
+            if exist_src or exist_bin:
+                correct_scanner_result(_output_dir, output_files, exist_src, exist_bin)
         if output_extension == ".xlsx":
             if remove_src_data:
                 overwrite_excel(_output_dir, default_oss_name, "OSS Name")
@@ -195,7 +214,14 @@ def run_scanner(src_path, dep_arguments, output_path, keep_raw_data=False,
             merge_yaml_files = [output_files["SRC"], output_files["BIN"], output_files["DEP"]]
             success, err_msg = merge_yamls(_output_dir, merge_yaml_files, final_report,
                                            remove_src_data, default_oss_name, url)
-
+        if correct_mode:
+            if exist_src:
+                shutil.move(os.path.join(_output_dir, tmp_dir, output_files['SRC']),
+                            os.path.join(_output_dir, output_files['SRC']))
+            if exist_bin:
+                shutil.move(os.path.join(_output_dir, tmp_dir, output_files['BIN']),
+                            os.path.join(_output_dir, output_files['BIN']))
+            shutil.rmtree(os.path.join(_output_dir, tmp_dir), ignore_errors=True)
         if success:
             if os.path.isfile(final_report):
                 result_log["Output File"] = final_report
@@ -212,7 +238,7 @@ def run_scanner(src_path, dep_arguments, output_path, keep_raw_data=False,
     try:
         if remove_src_data:
             logger.debug(f"Remove temporary source: {src_path}")
-            rmdir(src_path)
+            shutil.rmtree(src_path)
     except Exception as ex:
         logger.debug(f"Error to remove temp files:{ex}")
 
@@ -267,7 +293,7 @@ def init(output_path="", make_outdir=True):
 
 
 def run_main(mode, path_arg, dep_arguments, output_file_or_dir, file_format, url_to_analyze, db_url,
-             hide_progressbar=False, keep_raw_data=False, num_cores=-1):
+             hide_progressbar=False, keep_raw_data=False, num_cores=-1, correct_mode=True, correct_fpath=""):
     global _executed_path, _start_time
 
     output_file = ""
@@ -348,16 +374,20 @@ def run_main(mode, path_arg, dep_arguments, output_file_or_dir, file_format, url
                 default_oss_name = extract_name_from_link(url_to_analyze)
                 success, src_path = download_source(url_to_analyze, output_path)
 
+            if not correct_fpath:
+                correct_fpath = src_path
+
             if src_path != "":
                 run_scanner(src_path, dep_arguments, output_path, keep_raw_data,
                             run_src, run_bin, run_dep, run_prechecker,
                             remove_downloaded_source, {}, output_file,
                             output_extension, num_cores, db_url,
-                            default_oss_name, url_to_analyze)
+                            default_oss_name, url_to_analyze,
+                            correct_mode, correct_fpath)
         try:
             if not keep_raw_data:
                 logger.debug(f"Remove temporary files: {_output_dir}")
-                rmdir(_output_dir)
+                shutil.rmtree(_output_dir)
         except Exception as ex:
             logger.debug(f"Error to remove temp files:{ex}")
     except Exception as ex: