Merge pull request #195 from gue-ni/omicron_scripts

Add script for creating CSV files for vuln_batch_remediation.py

Author: mkumykov

examples/client/get_project_vulnerabilites_as_csv.py

@@ -0,0 +1,92 @@
+'''
+Export the vulnerabilites from a project as CSV. Can be used to apply batch vulnerability
+remediation with vuln_batch_remediation.py
+
+Output is in format:
+identifier, status, comment, componentName, componentVersion, description 
+
+The API token should be specified in a .env file or as environment variable.
+'''
+import re
+import os
+import sys
+import csv
+import logging
+import argparse
+from pprint import pprint
+from blackduck import Client
+from dotenv import load_dotenv
+
+load_dotenv()
+
+API_TOKEN = os.getenv('API_TOKEN')
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="[%(asctime)s] {%(module)s:%(lineno)d} %(levelname)s - %(message)s"
+)
+
+def strip_newline(str):
+    return str.replace('\r', '').replace('\n', ' ')
+
+def match_component(selected_components, component):
+    if (len(selected_components) == 0):
+        return True
+
+    for selected in selected_components:
+        if (re.search(selected, component, re.IGNORECASE)):
+            return True
+
+    return False
+
+def main():
+    program_name = os.path.basename(sys.argv[0])
+    parser = argparse.ArgumentParser(prog=program_name, usage="%(prog)s [options]", description="Automated Assessment")
+    parser.add_argument("project", help="project name")
+    parser.add_argument("version",  help="project version, e.g. latest")
+    parser.add_argument("--output", required=False,help="csv output path" )
+    parser.add_argument("--base-url", required=False, help="base url", default="https://blackduck.omicron.at")
+    parser.add_argument("--components", required=False, help="component names, comma seperated without space")
+    args = parser.parse_args()
+
+    components = args.components.split(',') if args.components != None else []
+    projectname = args.project
+    projectversion = args.version
+    output = args.output  if  args.output != None else "output.csv"
+
+    csv_file = open(output, mode='w', newline='', encoding='utf-8')
+    csv_writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
+
+    bd = Client(
+        token=API_TOKEN,
+        base_url=args.base_url,
+        verify=False  # TLS certificate verification
+    )
+
+    for project in bd.get_resource('projects'):
+        if (project['name'] == projectname):
+            for version in bd.get_resource('versions', project):
+                if (projectversion == None):
+                    pprint(version['versionName'])
+
+                else:
+                    if (version['versionName'] == projectversion):
+                        for vulnverable_component in bd.get_resource('vulnerable-components', version):
+                            componentName = vulnverable_component["componentName"]
+
+                            if (match_component(components, componentName)):
+                                componentVersion = vulnverable_component["componentVersionName"]
+                                remediation = vulnverable_component['vulnerabilityWithRemediation']
+
+                                status = remediation['remediationStatus']
+                                identifier = remediation['vulnerabilityName']
+                                description = strip_newline(remediation['description'])
+                                comment = strip_newline(remediation.get('remediationComment', ""))
+                                
+                                row =  [identifier, status, comment, componentName, componentVersion, description]
+                                csv_writer.writerow(row)
+                        break
+            break
+
+if __name__ == "__main__":
+    main()

examples/vuln_batch_remediation.py

@@ -26,11 +26,13 @@
 Each processing step can be turned on or off.  At least one step must be run.  Default
 is to run both.
 
-The script get's it CVE and orign lists from CSV files.  The CSV filenames are loaded
+The script can get's its CVE and orign lists from CSV files.  The CSV filenames are loaded
 from Custom Fields in the Black Duck project.  This allows different groups of projects to
 use different remeidation settings.  If a CVE remediation status should apply globally
 to all projects, Black Duck's global remediation feature should be used.
 
+The script can also get the CSV filenames from the command line arguments.
+
 Here is an example of the CSV data for the CVE list:
 
 "CVE-2016-1840","IGNORED","Applies only to Apple OS"
@@ -67,7 +69,7 @@
 import json
 import csv
 import traceback
-
+from pprint import pprint
 from argparse import ArgumentParser
 from argparse import RawDescriptionHelpFormatter
 
@@ -81,15 +83,20 @@
 
 
 def load_remediation_input(remediation_file):
-    with open(remediation_file, mode='r') as infile:
+    with open(remediation_file, mode='r', encoding="utf-8") as infile:
         reader = csv.reader(infile)
-        return {rows[0]:[rows[1],rows[2]] for rows in reader}
+        #return {rows[0]:[rows[1],rows[2]] for rows in reader}
+        return {rows[0]:rows[1:] for rows in reader}
 
 def remediation_is_valid(vuln, remediation_data):
     vulnerability_name = vuln['vulnerabilityWithRemediation']['vulnerabilityName']
-    # remediation_status = vuln['vulnerabilityWithRemediation']['remediationStatus']
-    # remediation_comment = vuln['vulnerabilityWithRemediation'].get('remediationComment','')
+    remediation_status = vuln['vulnerabilityWithRemediation']['remediationStatus']
+    remediation_comment = vuln['vulnerabilityWithRemediation'].get('remediationComment','')
+    
     if vulnerability_name in remediation_data.keys():
+        remediation = remediation_data[vulnerability_name]
+        if (remediation_status == remediation[0] and remediation_comment == remediation[1]):
+            return None
         return remediation_data[vulnerability_name]
     else:
         return None
@@ -114,12 +121,31 @@ def find_custom_field_value (custom_fields, custom_field_label):
                 return None
     return None
 
-def process_vulnerabilities(hub, vulnerable_components, remediation_data=None, exclusion_data=None):
+
+
+def set_vulnerablity_remediation(hub, vuln, remediation_status, remediation_comment):
+    url = vuln['_meta']['href']
+    update={}
+    update['remediationStatus'] = remediation_status
+    update['comment'] = remediation_comment
+    response = hub.execute_put(url, data=update)
+    return response
+
+def process_vulnerabilities(hub, vulnerable_components, remediation_data=None, exclusion_data=None, dry_run=False):
+
+    if (dry_run):
+        print(f"Opening dry run output file: {dry_run}")
+        csv_file = open(dry_run, mode='w', newline='', encoding='utf-8')
+        csv_writer = csv.writer(csv_file, delimiter=',', quotechar='"', quoting=csv.QUOTE_MINIMAL)
+
     count = 0
-    print('"Component Name","Component Version","Component OriginID","CVE","Reason","Remeidation Status","HTTP response code"')
+    print('"Component Name","Component Version","CVE","Reason","Remeidation Status","HTTP response code"')
 
     for vuln in vulnerable_components['items']:
         if vuln['vulnerabilityWithRemediation']['remediationStatus'] == "NEW":
+            remediation_action = None 
+            exclusion_action = None
+
             if (remediation_data):
                 remediation_action = remediation_is_valid(vuln, remediation_data)
 
@@ -137,14 +163,20 @@ def process_vulnerabilities(hub, vulnerable_components, remediation_data=None, e
                 reason = 'origin-exclusion'
 
             if (remediation_action):
-                resp = hub.set_vulnerablity_remediation(vuln, remediation_action[0],remediation_action[1])
-                count += 1
-                print ('\"{}\",\"{}\",\"{}\",\"{}\",\"{}\",\"{}\",\"{}\"'.
+                if (dry_run):
+                    remediation_action.insert(0, vuln['vulnerabilityWithRemediation']['vulnerabilityName'])
+                    csv_writer.writerow(remediation_action)
+                else:
+                    resp = set_vulnerablity_remediation(hub, vuln, remediation_action[0],remediation_action[1])
+                    count += 1
+                
+                print ('\"{}\",\"{}\",\"{}\",\"{}\",\"{}\",\"{}\"'.
                     format(vuln['componentName'], vuln['componentVersionName'],
-                    vuln['componentVersionOriginId'], 
                     vuln['vulnerabilityWithRemediation']['vulnerabilityName'],
-                    reason, remediation_action[0], resp.status_code))
-    print (f'Remediated {count} vulnerabilities.')
+                    reason, remediation_action[0],  resp.status_code if not dry_run else ""))
+               
+
+    print (f'Remediated {count} vulnerabilities. {"(dry run)" if dry_run else ""}')
 
 def main(argv=None): # IGNORE:C0111
     '''Command line options.'''
@@ -157,7 +189,7 @@ def main(argv=None): # IGNORE:C0111
     program_name = os.path.basename(sys.argv[0])
     program_version = "v%s" % __version__
     program_build_date = str(__updated__)
-    program_version_message = '%%(prog)s %s (%s)' % (program_version, program_build_date)
+    program_version_message = '%s %s (%s)' % (program_name, program_version, program_build_date)
     program_shortdesc = __import__('__main__').__doc__.split("\n")[1]
     program_license = '''%s
 
@@ -178,8 +210,11 @@ def main(argv=None): # IGNORE:C0111
         parser = ArgumentParser(description=program_license, formatter_class=RawDescriptionHelpFormatter)
         parser.add_argument("projectname", help="Project nname")
         parser.add_argument("projectversion", help="Project vesrsion")
-        parser.add_argument("--no-process-cve-remediation-list", dest='process_cve_remediation_list', action='store_false', help="Disbable processing CVE-Remediation-list")
-        parser.add_argument("--no-process-origin-exclusion-list", dest='process_origin_exclusion_list', action='store_false', help="Disable processing Origin-Exclusion-List")
+        parser.add_argument("--dry-run", dest="dry_run", nargs='?', const="dry_run.csv", help="dry run remediations and output to file")
+        parser.add_argument("--remediation-list", dest="local_remediation_list", default=None, help="Filename of cve remediation list csv file")
+        parser.add_argument("--origin-exclusion-list", dest="local_origin_exclusion_list", default=None, help="Filename of origin exclusion list csv file")
+        parser.add_argument("--no-process-cve-remediation-list", dest='process_cve_remediation_list', action='store_false', help="Disable processing CVE-Remediation-list")
+        parser.add_argument("--no-process-origin-exclusion-list", dest='process_origin_exclusion_list', default=None, action='store_false', help="Disable processing Origin-Exclusion-List")
         parser.add_argument("--cve-remediation-list-custom-field-label", default='CVE Remediation List', help='Label of Custom Field on Black Duck that contains remeidation list file name')
         parser.add_argument("--origin-exclusion-list-custom-field-label", default='Origin Exclusion List', help='Label of Custom Field on Black Duck that containts origin exclusion list file name')
         parser.add_argument('-V', '--version', action='version', version=program_version_message)
@@ -189,9 +224,15 @@ def main(argv=None): # IGNORE:C0111
 
         projectname = args.projectname
         projectversion = args.projectversion
+        local_cve_remediation_file = args.local_remediation_list
+        local_origin_exclusion_file = args.local_origin_exclusion_list
         process_cve_remediation = args.process_cve_remediation_list
         process_origin_exclulsion = args.process_origin_exclusion_list
-       
+        #dry_run = args.dry_run
+        #dry_run_output = args.dry_run_output
+        dry_run = args.dry_run
+        print(args.dry_run)
+
         message = f"{program_version_message}\n\n Project: {projectname}\n Version: {projectversion}\n Process origin exclusion list: {process_origin_exclulsion}\n Process CVE remediation list: {process_cve_remediation}"
         print (message)
 
@@ -203,26 +244,37 @@ def main(argv=None): # IGNORE:C0111
         hub = HubInstance()
         project = hub.get_project_by_name(projectname)
         version = hub.get_project_version_by_name(projectname, projectversion)
-        custom_fields = hub.get_project_custom_fields (project)
+        
+        custom_fields = hub.get_cf_values(project)
 
         if (process_cve_remediation):
-            cve_remediation_file = find_custom_field_value (custom_fields, args.cve_remediation_list_custom_field_label)
-            print (f' Opening: {args.cve_remediation_list_custom_field_label}:{cve_remediation_file}')
+            if (local_cve_remediation_file):
+                cve_remediation_file = local_cve_remediation_file
+                print (f' Opening CVE remediation file: {cve_remediation_file}')
+            else:
+                cve_remediation_file = find_custom_field_value (custom_fields, args.cve_remediation_list_custom_field_label)
+                print (f' Opening: {args.cve_remediation_list_custom_field_label}:{cve_remediation_file}')
+
             remediation_data = load_remediation_input(cve_remediation_file)
         else:
             remediation_data = None
 
         if (process_origin_exclulsion):
-            exclusion_list_file = find_custom_field_value (custom_fields, args.origin_exclusion_list_custom_field_label)
-            print (f' Opening: {args.origin_exclusion_list_custom_field_label}:{exclusion_list_file}')
+            if local_origin_exclusion_file:
+                exclusion_list_file = local_origin_exclusion_file
+                print (f' Opening origin exclusion list: {exclusion_list_file}')
+            else:
+                exclusion_list_file = find_custom_field_value (custom_fields, args.origin_exclusion_list_custom_field_label)
+                print (f' Opening: {args.origin_exclusion_list_custom_field_label}:{exclusion_list_file}')
             exclusion_data = load_remediation_input(exclusion_list_file)
         else:
             exclusion_data = None
+    
 
         # Retrieve the vulnerabiltites for the project version
         vulnerable_components = hub.get_vulnerable_bom_components(version)
 
-        process_vulnerabilities(hub, vulnerable_components, remediation_data, exclusion_data)
+        process_vulnerabilities(hub, vulnerable_components, remediation_data, exclusion_data, dry_run)
 
         return 0
     except Exception: