Adding support for retrieving vulnerability info for BOM components

blackduck/HubRestApi.py

@@ -213,6 +213,12 @@ def get_link(self, bd_rest_obj, link_name):
  else:
  logging.debug("This does not appear to be a BD REST object. It should have ['_meta']['links']")
 
+ def get_limit_paramstring(self, limit):
+ return "?limit={}".format(limit)
+
+ def get_apibase(self):
+ return self.config['baseurl'] + "/api"
+
  ###
  #
  # Role stuff
@@ -484,26 +490,18 @@ def find_component_info_for_protex_component(self, protex_component_id, protex_c
  else:
  return component_list_d['items']
 
- def get_limit_paramstring(self, limit):
- return "?limit={}".format(limit)
-
- def get_apibase(self):
- return self.config['baseurl'] + "/api"
-
- def get_version_by_name(self, project, version_name):
- version_list = self.get_project_versions(project)
- for version in version_list['items']:
- if version['versionName'] == version_name:
- return version
-
- def _get_version_link(self, version, link_type):
- # if link_type == 'licenseReports':
- # version_id = version['_meta']['href'].split("/")[-1]
- # return self.get_urlbase() + "/api/v1/versions/{}/reports".format(version_id)
- # else:
- for link in version['_meta']['links']:
- if link['rel'] == link_type:
- return link['href']
+ def get_vulnerable_bom_components(self, version_obj, limit=9999):
+ url = "{}/vulnerable-bom-components".format(version_obj['_meta']['href'])
+ custom_headers = {'Content-Type': 'application/vnd.blackducksoftware.bill-of-materials-4+json'}
+ param_string = self._get_parameter_string({'limit': limit})
+ url = "{}{}".format(url, param_string)
+ response = self.execute_get(url, custom_headers=custom_headers)
+ if response.status_code == 200:
+ vulnerable_bom_components = response.json()
+ return vulnerable_bom_components
+ else:
+ logging.warning("Failed to retrieve vulnerable bom components for project {}, status code {}".format(
+ version_obj, response.status_code))
 
  ##
  #
@@ -524,7 +522,7 @@ def create_version_reports(self, version, report_list, format="CSV"):
  'reportType': 'VERSION',
  'reportFormat': format
  }
- version_reports_url = self._get_version_link(version, 'versionReport')
+ version_reports_url = self.get_link(version, 'versionReport')
  return self.execute_post(version_reports_url, post_data)
 
  valid_notices_formats = ["TEXT", "HTML"]
@@ -537,7 +535,7 @@ def create_version_notices_report(self, version, format="TEXT"):
  'reportType': 'VERSION_LICENSE',
  'reportFormat': format
  }
- notices_report_url = self._get_version_link(version, 'licenseReports')
+ notices_report_url = self.get_link(version, 'licenseReports')
  return self.execute_post(notices_report_url, post_data)
 
  def download_report(self, report_id):
@@ -788,20 +786,22 @@ def get_project_by_name(self, project_name):
  if project['name'] == project_name:
  return project
 
+ def get_version_by_name(self, project, version_name):
+ version_list = self.get_project_versions(project, parameters={'q':"versionName:{}".format(version_name)})
+ # A query by name can return more than one version if other versions
+ # have names that include the search term as part of their name
+ for version in version_list['items']:
+ if version['versionName'] == version_name:
+ return version
+
  def get_project_version_by_name(self, project_name, version_name):
  project = self.get_project_by_name(project_name)
  if project:
- project_versions = self.get_project_versions(
- project, 
- parameters={'q':"versionName:{}".format(version_name)}
- )
- # A query by name can return more than one version if other versions
- # have names that include the search term as part of their name
- for project_version in project_versions['items']:
- if project_version['versionName'] == version_name:
- logging.debug("Found matching version: {}".format(project_version))
- return project_version
- logging.debug("Did not find any project version matching {}".format(version_name))
+ version = self.get_version_by_name(project, version_name)
+ if version == None:
+ logging.debug("Did not find any project version matching {}".format(version_name))
+ else:
+ return version
  else:
  logging.debug("Did not find a project with name {}".format(project_name))
 

blackduck/__version__.py

@@ -1,4 +1,4 @@
 
-VERSION = (0, 0, 14)
+VERSION = (0, 0, 15)
 
 __version__ = '.'.join(map(str, VERSION))

examples/get_bom_component_vulnerability_info.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python
+
+import argparse
+from datetime import datetime
+import json
+import logging
+import sys
+import timestring
+
+from blackduck.HubRestApi import HubInstance
+
+#
+# Example usage:
+#	To get all the vulnerabilities (first time) and save the last run date/time,
+#	python examples/get_bom_component_vulnerability_info.py struts2-showcase 2.6-SNAPSHOT -s > vulnerabilities.json
+#
+#	Having saved the last run date/time, use it to view any newly published vulnerabilities that have come out since the last run,
+#	python examples/get_bom_component_vulnerability_info.py struts2-showcase 2.6-SNAPSHOT -n `cat .last_run` > vulnerabilities_since.json
+#
+#	Having saved the last run date/time, use it to view any newly published vulnerabilities that have come out since the last run
+#	and update the last run date/time,
+#	python examples/get_bom_component_vulnerability_info.py struts2-showcase 2.6-SNAPSHOT -s -n `cat .last_run` > vulnerabilities_since.json
+#
+#	Use --newer_than (aka -n) to specify your own date (or date/time),
+#	python examples/get_bom_component_vulnerability_info.py struts2-showcase 2.6-SNAPSHOT -n "2017" > vulnerabilities_since_2017.json
+#	python examples/get_bom_component_vulnerability_info.py struts2-showcase 2.6-SNAPSHOT -n "July 1 2018" > vulnerabilities_since_July_1_2018.json
+#	python examples/get_bom_component_vulnerability_info.py struts2-showcase 2.6-SNAPSHOT -n "July 1 2018 5:30 pm" > vulnerabilities_since_July_1_2018_1730.json
+
+parser = argparse.ArgumentParser("Retreive BOM component vulnerability information for the given project and version")
+parser.add_argument("project_name")
+parser.add_argument("version")
+parser.add_argument("-n", "--newer_than", 
+default=None, 
+type=str,
+help="Set this option to see all vulnerabilities published since the given date/time.")
+parser.add_argument("-s", "--save_dt", 
+action='store_true', 
+help="If set, the date/time will be saved to a file named '.last_run' in the current directory which can be used later with the -n option to see vulnerabilities published since the last run.")
+args = parser.parse_args()
+
+if args.newer_than:
+newer_than = timestring.Date(args.newer_than).date
+else:
+newer_than = None
+
+if args.save_dt:
+with open(".last_run", "w") as f:
+f.write(datetime.now().isoformat())
+
+logging.basicConfig(format='%(asctime)s:%(levelname)s:%(message)s', stream=sys.stderr, level=logging.DEBUG)
+logging.getLogger("requests").setLevel(logging.WARNING)
+logging.getLogger("urllib3").setLevel(logging.WARNING)
+
+hub = HubInstance()
+
+project = hub.get_project_by_name(args.project_name)
+
+version = hub.get_version_by_name(project, args.version)
+
+vulnerable_bom_components_info = hub.get_vulnerable_bom_components(version)
+
+vulnerable_bom_components = vulnerable_bom_components_info.get('items', [])
+
+if vulnerable_bom_components:
+vulnerable_bom_components = sorted(
+vulnerable_bom_components, 
+key = lambda k: k['vulnerabilityWithRemediation']['vulnerabilityPublishedDate'])
+if newer_than:
+vulnerable_bom_components = [v for v in vulnerable_bom_components 
+if timestring.Date(v['vulnerabilityWithRemediation']['vulnerabilityPublishedDate']) > newer_than ]
+else:
+logging.debug("Did not find any vulnerable BOM components in project {}, version {}".format(args.project_name, args.version))
+
+print(json.dumps(vulnerable_bom_components))
+

requirements.txt

@@ -4,6 +4,7 @@ python-dateutil>=2.8.0
 
 # for examples printing tables to the terminal
 terminaltables
+timestring
 
 # for unit testing
 pytest

test/test_hub_rest_api_python.py

@@ -402,22 +402,6 @@ def test_create_version_reports(requests_mock, mock_hub_instance):
 def test_create_version_notices_report(requests_mock, mock_hub_instance):
  pass
 
-def test_get_version_link(mock_hub_instance):
- version_json = json.load(open("version.json"))
-
- # replace the base URL with the mocked base url in all the links
- baseurl = mock_hub_instance.get_urlbase()
- for link_d in version_json['_meta']['links']:
- link_d['href'] = re.sub("https://.*/api", "{}/api".format(baseurl), link_d['href'])
-
- for link_name in ['versionReport', 'licenseReports', 'riskProfile', 'components', 'vulnerable-components', 'comparison', 'project', 'policy-status', 'codelocations']:
- url = mock_hub_instance._get_version_link(version_json, link_name)
- assert url # we got something
-
- parsed_url = urlparse(url)
- assert parsed_url.scheme == urlparse(fake_hub_host).scheme
- assert parsed_url.netloc == urlparse(fake_hub_host).netloc
-
 @pytest.fixture()
 def unreviewed_snippet_json():
  with open("unreviewed_snippet.json") as f: