A program that creates a csv containing vulnerability remediation audit info for a single project or all versions within a project

Author: evansa-synopsys

examples/generate_vulnerability_remediation_info.py

@@ -0,0 +1,167 @@
+import argparse
+import csv
+import glob
+import logging
+import os
+import shutil
+import sys
+import time
+import timeit
+
+import pandas
+from pandas.errors import EmptyDataError
+
+from blackduck.HubRestApi import HubInstance
+
+parser = argparse.ArgumentParser(
+    "A program that creates a csv containing vulnerability remediation audit info for a specified project version")
+parser.add_argument("project")
+parser.add_argument('-V', '--version', default="" )
+parser.add_argument('-r', '--refresh', action='store_true',
+                    help='delete existing reports in the results directory and regenerate')
+parser.add_argument('-v', '--verbose', action='store_true', default=False, help='turn on DEBUG logging')
+
+args = parser.parse_args()
+
+
+def set_logging_level(log_level):
+    logging.basicConfig(stream=sys.stderr, level=log_level)
+
+
+if args.verbose:
+    set_logging_level(logging.DEBUG)
+else:
+    set_logging_level(logging.INFO)
+
+projname = args.project
+timestamp = time.strftime('%m_%d_%Y_%H_%M')
+file_out = (projname + '_' + "vulnerability_remediation_report-" + timestamp)
+file_out = (file_out + ".csv")
+hub = HubInstance()
+rootDir = os.getcwd()
+
+
+def do_refresh(dir_name):
+    temp_dir = os.path.join(rootDir, dir_name)
+    print("tempDir=%s" % temp_dir)
+    for fileName in os.listdir(temp_dir):
+        print("Removing stale files %s" % fileName)
+        os.remove(os.path.join(temp_dir, fileName))
+
+
+def check_dirs():
+    os.chdir(rootDir)
+    if not os.path.isdir('./temp'):
+        os.makedirs('./temp')
+        print('made temp directory')
+    elif len(os.listdir('./temp')) != 0:
+        do_refresh('temp')
+    else:
+        print('temp directory already exists')
+
+    if not os.path.isdir('./vulnerability_remediation_results'):
+        os.makedirs('./vulnerability_remediation_results')
+        print('made vulnerability_remediation_results directory')
+    elif args.refresh and len(os.listdir('./vulnerability_remediation_results')) != 0:
+        print('refreshing vulnerability_remediation_results')
+        do_refresh('vulnerability_remediation')
+    else:
+        print('vulnerability_remediation directory already exists')
+
+
+# The resultant report will contain these columns for each vulnerability
+def get_header():
+    return ["Project Name", "Project Version Name", "Component Name", "Component Version",
+            "Vulnerability ID", "Remediation Status", "Created By",
+            "Updated By", "Remediation Comment", "Created At", "Updated At"]
+
+
+def genreport():
+    project = hub.get_project_by_name(args.project)
+    vulnerable_bom_components_all_versions = dict()
+    if args.version:
+        version = hub.get_version_by_name(project, args.version)
+        project_version_key = "{}, {}".format(project['name'], version['versionName'])
+        version_dict = hub.get_vulnerable_bom_components(version)
+        vulnerable_bom_components_all_versions.update(
+            {project_version_key: {"total_count": version_dict['totalCount'],
+                                   "vulnerable_components": version_dict['items']}})
+        print("Vulnerability count returned for {} {} = {} ".format(project['name'], version['versionName'],
+                                                                    version_dict['totalCount']))
+    else:
+        versions = hub.get_project_versions(project)['items']
+        for v in versions:
+            project_version_key = "{}, {}".format(project['name'], v['versionName'])
+            version_dict = hub.get_vulnerable_bom_components(v)
+            vulnerable_bom_components_all_versions.update(
+                {project_version_key: {"total_count": version_dict['totalCount'],
+                                       "vulnerable_components": version_dict['items']}})
+            print("Vulnerability count returned for {} {} = {} ".format(project['name'], v['versionName'],
+                                                                        version_dict['totalCount']))
+
+    curdir = os.getcwd()
+    tempdir = os.path.join(curdir, 'temp')
+    os.chdir(tempdir)
+    with open(file_out, 'a', newline='') as f:
+        writer = csv.writer(f)
+        first_file = True
+
+        for version in vulnerable_bom_components_all_versions:
+            for bom_component in vulnerable_bom_components_all_versions.get(version).get('vulnerable_components'):
+                # retrieve the remediation audit trail for a vulnerability
+                project_name = version.split(',')[0]
+                project_version_name = version.split(',')[1]
+                component_name = bom_component['componentName']
+                component_version_name = bom_component['componentVersionName']
+
+                if first_file:
+                    header = get_header()
+                    writer.writerow(header)
+                    first_file = False
+                try:
+                    dates = list(
+                        map(lambda d: d.split('T')[0], [bom_component['vulnerabilityWithRemediation'].get(y) for y in
+                                                        ['remediationCreatedAt',
+                                                         'remediationUpdatedAt']]))
+                    row_list = [project_name, project_version_name, component_name, component_version_name] + \
+                               [bom_component['vulnerabilityWithRemediation'].get(x) for x in
+                                ['vulnerabilityName',
+                                 'remediationStatus',
+                                 'remediationCreatedBy',
+                                 'remediationUpdatedBy',
+                                 'remediationComment']] + dates
+
+                except (KeyError, IndexError) as err:
+                    logging.debug("Issue pulling vulnerability remediation for:{}, {} not present".format(
+                        bom_component['vulnerabilityWithRemediation']['vulnerabilityName'], err))
+                else:
+                    writer.writerow(row_list)
+
+
+def concat():
+    curdir = os.getcwd()
+    os.chdir(curdir)
+    all_csvs = glob.glob(os.path.join(curdir, '*.csv'))
+    all_data_frames = []
+    for a_csv in all_csvs:
+        try:
+            data_frame = pandas.read_csv(a_csv, index_col=None)
+        except EmptyDataError:
+            data_frame = pandas.DataFrame()
+
+        all_data_frames.append(data_frame)
+    data_frame_concat = pandas.concat(all_data_frames, axis=0, ignore_index=True)
+    data_frame_concat.to_csv(file_out, index=False)
+    shutil.move(file_out, '../vulnerability_remediation_results/')
+    shutil.rmtree('../temp', ignore_errors=True)
+
+
+def main():
+    check_dirs()
+    start = timeit.default_timer()
+    genreport()
+    print("Time spent generating this report: {} seconds".format(int(timeit.default_timer() - start)))
+    concat()
+
+
+main()