adding utilities to dump, upload users, user groups, policies

Author: Glenn Snyder

blackduck/HubRestApi.py

@@ -175,6 +175,141 @@ def _get_parameter_string(self, parameters={}):
  parameter_string = "&".join(["{}={}".format(k,v) for k,v in parameters.items()])
  return "?" + parameter_string
 
+ ###
+ #
+ # Role stuff
+ #
+ ###
+ def _get_role_url(self):
+ return self.config['baseurl'] + "/api/roles"
+
+ def get_roles(self, parameters={}):
+ url = self._get_role_url() + self._get_parameter_string(parameters)
+ response = self.execute_get(url)
+ return response.json()
+
+ def get_roles_url_from_user_or_group(self, user_or_group):
+ roles_url = None
+ for endpoint in user_or_group['_meta']['links']:
+ if endpoint['rel'] == "roles":
+ roles_url = endpoint['href']
+ return roles_url
+
+ def get_roles_for_user_or_group(self, user_or_group):
+ roles_url = self.get_roles_url_from_user_or_group(user_or_group)
+ if roles_url:
+ return self.execute_get(roles_url)
+ else:
+ return None
+
+ def get_role_url_by_name(self, role_name):
+ # Return the role URL for this server corresponding to the role name
+ all_roles = self.get_roles()
+ for role in all_roles['items']:
+ if role['name'] == role_name:
+ return role['_meta']['href']
+ return None
+
+ def assign_role_to_user_or_group(self, role_name, user_or_group):
+ user_or_group_roles_url = self.get_roles_url_from_user_or_group(user_or_group)
+ return self.assign_role_given_role_url(role_name, user_or_group_roles_url)
+
+ def assign_role_given_role_url(self, role_name, user_or_group_role_assignment_url):
+ role_url = self.get_role_url_by_name(role_name)
+ if self.bd_major_version == "3":
+ # A hack to get the assignment to work on v3
+ role_url = role_url.replace("api", "api/internal")
+ data = {"name": role_name, "role": role_url}
+ logging.debug("executing POST to {} with {}".format(
+ user_or_group_role_assignment_url, data))
+ return self.execute_post(user_or_group_role_assignment_url, data = data)
+
+
+ ###
+ #
+ # User stuff
+ #
+ ###
+ def _get_user_url(self):
+ return self.config['baseurl'] + "/api/users"
+
+ def get_users(self, parameters={}):
+ url = self._get_user_url() + self._get_parameter_string(parameters)
+ response = self.execute_get(url)
+ return response.json()
+
+ def create_user(self, user_json):
+ url = self._get_user_url()
+ location = self._create(url, user_json)
+ return location
+
+ def get_user_by_id(self, user_id):
+ url = self._get_user_url() + "/{}".format(user_id)
+ return self.get_user_by_url(url)
+
+ def get_user_by_url(self, user_url):
+ response = self.execute_get(user_url)
+ jsondata = response.json()
+ return jsondata
+
+ def update_user_by_id(self, user_id, update_json):
+ url = self._get_user_url() + "/{}".format(user_id)
+ return self.update_user_by_url(url, update_json)
+
+ def update_user_by_url(self, user_url, update_json):
+ return self.execute_put(user_url, update_json)
+
+ def delete_user_by_id(self, user_id):
+ url = self._get_user_url() + "/{}".format(user_id)
+ return self.delete_user_by_url(url)
+
+ def delete_user_by_url(self, user_url):
+ return self.execute_delete(user_url)
+
+ ###
+ #
+ # User group stuff
+ #
+ ###
+ def _get_user_group_url(self):
+ return self.config['baseurl'] + "/api/usergroups"
+
+ def get_user_groups(self, parameters={}):
+ url = self._get_user_group_url() + self._get_parameter_string(parameters)
+ response = self.execute_get(url)
+ return response.json()
+
+ def create_user_group(self, user_group_json):
+ if self.bd_major_version == "3":
+ url = self.config['baseurl'] + '/api/v1/usergroups'
+ else:
+ url = self._get_user_group_url()
+ location = self._create(url, user_group_json)
+ return location
+
+ def get_user_group_by_id(self, user_group_id):
+ url = self._get_user_group_url() + "/{}".format(user_group_id)
+ return self.get_user_group_by_url(url)
+
+ def get_user_group_by_url(self, user_group_url):
+ response = self.execute_get(user_group_url)
+ jsondata = response.json()
+ return jsondata
+
+ def update_user_group_by_id(self, user_group_id, update_json):
+ url = self._get_user_group_url() + "/{}".format(user_group_id)
+ return self.update_user_group_by_url(url, update_json)
+
+ def update_user_group_by_url(self, user_group_url, update_json):
+ return self.execute_put(user_group_url, update_json)
+
+ def delete_user_group_by_id(self, user_group_id):
+ url = self._get_user_group_url() + "/{}".format(user_group_id)
+ return self.delete_user_group_by_url(url)
+
+ def delete_user_group_by_url(self, user_group_url):
+ return self.execute_delete(user_group_url)
+
  ###
  #
  # Policy stuff
@@ -527,8 +662,19 @@ def execute_put(self, url, data):
 
  def _create(self, url, json_body):
  response = self.execute_post(url, json_body)
- if response.status_code == 201 and "location" in response.headers:
- return (response.headers["location"])
+ response_json = response.json()
+ # v4+ returns the newly created location in the response headers
+ # and there is nothing in the response json
+ # whereas v3 returns the newly created object in the response json
+ if response.status_code == 201:
+ if "location" in response.headers:
+ return (response.headers["location"])
+ elif '_meta' in response_json and 'href' in response_json['_meta']:
+ return response_json['_meta']['href']
+ else:
+ # worst case, try returning the whole response json and let
+ # client figure it out
+ return response.json()
  elif response.status_code == 412:
  raise CreateFailedAlreadyExists("Failed to create the object because it already exists - url {}, body {}, response {}".format(url, json_body, response))
  else:

examples/dump_policies.py

@@ -0,0 +1,73 @@
+'''
+Created on Dec 5, 2018
+
+@author: gsnyder
+
+Dump policies to a file
+
+'''
+import argparse
+import json
+import logging
+from pprint import pprint
+import sys
+
+from blackduck.HubRestApi import HubInstance, CreateFailedAlreadyExists
+
+
+def serialize_to_file(file, data):
+with open(file, "w") as f:
+json.dump(data, f, indent=3)
+
+parser = argparse.ArgumentParser("Dump policies from a Hub instance")
+parser.add_argument("base_filename", help="There will be two files dumped - one raw and one prepped to copy policies to another Hub instance")
+args = parser.parse_args()
+
+logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
+
+source_hub = HubInstance()
+
+policy_keys_to_delete = [
+'_meta', 
+'href', 
+'links', 
+'createdAt', 
+'createdBy',
+'createdByUser',
+'updatedAt', 
+'updatedBy',
+'updatedByUser']
+
+policies_to_copy = source_hub.get_policies()
+
+if 'totalCount' in policies_to_copy and policies_to_copy['totalCount'] > 0:
+logging.debug("Dumping {} policies from {}".format(policies_to_copy['totalCount'], source_hub.config['baseurl']))
+assert 'items' in policies_to_copy, "Should always be a list of items in a non-zero list of policies"
+
+policy_data_raw = []
+policy_data_prepped_to_copy = []
+
+for policy in policies_to_copy['items']:
+import pdb; pdb.set_trace()
+policy = source_hub.get_policy_by_url(policy['_meta']['href'])
+policy_data_raw.append(policy)
+
+policy_copy_prepped_to_create = policy.copy()
+
+for k in policy_keys_to_delete:
+if k in policy_copy_prepped_to_create:
+del policy_copy_prepped_to_create[k]
+
+policy_data_prepped_to_copy.append(policy_copy_prepped_to_create)
+
+raw_file = args.base_filename + ".raw"
+prepped_file = args.base_filename + ".prepped"
+
+serialize_to_file(raw_file, policy_data_raw)
+serialize_to_file(prepped_file, policy_data_prepped_to_copy)
+else:
+logging.info("No policies to dump")
+
+
+
+

examples/dump_user_groups.py

@@ -0,0 +1,73 @@
+'''
+Created on Dec 5, 2018
+
+@author: gsnyder
+
+Dump user groups to a file
+
+'''
+import argparse
+import json
+import logging
+from pprint import pprint
+import sys
+
+from blackduck.HubRestApi import HubInstance, CreateFailedAlreadyExists
+
+
+def serialize_to_file(file, data):
+with open(file, "w") as f:
+json.dump(data, f, indent=3)
+
+parser = argparse.ArgumentParser("Dump user_groups from a Hub instance")
+parser.add_argument("base_filename", help="There will be two files dumped - one raw and one prepped to copy user_groups to another Hub instance")
+args = parser.parse_args()
+
+logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
+
+source_hub = HubInstance()
+
+user_group_keys_to_delete = ['_meta', 'href', 'links']
+
+user_groups_to_copy = source_hub.get_user_groups()
+
+if 'totalCount' in user_groups_to_copy and user_groups_to_copy['totalCount'] > 0:
+logging.debug("Dumping {} user_groups from {}".format(user_groups_to_copy['totalCount'], source_hub.config['baseurl']))
+assert 'items' in user_groups_to_copy, "Should always be a list of items in a non-zero list of user_groups"
+
+user_group_data_raw = []
+user_group_data_prepped_to_copy = []
+
+for user_group in user_groups_to_copy['items']:
+user_group_roles_response = source_hub.get_roles_for_user_or_group(user_group)
+if user_group_roles_response:
+user_group_roles = user_group_roles_response.json()
+else:
+user_group_roles = []
+
+user_group_data_raw.append({
+'user_group': user_group, 'roles': user_group_roles
+})
+
+user_group_copy_prepped_to_create = user_group.copy()
+for k in user_group_keys_to_delete:
+if k in user_group_copy_prepped_to_create:
+del user_group_copy_prepped_to_create[k]
+
+user_group_roles_to_create = [role['name'] for role in user_group_roles['items']]
+
+user_group_data_prepped_to_copy.append({
+'user_group': user_group_copy_prepped_to_create, 'roles': user_group_roles_to_create
+})
+
+raw_file = args.base_filename + ".raw"
+prepped_file = args.base_filename + ".prepped"
+
+serialize_to_file(raw_file, user_group_data_raw)
+serialize_to_file(prepped_file, user_group_data_prepped_to_copy)
+else:
+logging.info("No user groups to dump")
+
+
+
+

examples/dump_users.py

@@ -0,0 +1,80 @@
+'''
+Created on Dec 4, 2018
+
+@author: gsnyder
+
+Copy users and their roles from one Hub instance to another
+
+'''
+import argparse
+import json
+import logging
+from pprint import pprint
+import sys
+
+from blackduck.HubRestApi import HubInstance, CreateFailedAlreadyExists
+
+
+def serialize_to_file(file, data):
+with open(file, "w") as f:
+json.dump(data, f, indent=3)
+
+parser = argparse.ArgumentParser("Dump users from a Hub instance")
+parser.add_argument("base_filename", help="There will be two files dumped - one raw and one prepped to copy users to another Hub instance")
+args = parser.parse_args()
+
+logging.basicConfig(stream=sys.stdout, level=logging.DEBUG)
+
+source_hub = HubInstance()
+
+user_keys_to_delete = ['_meta', 'href', 'links']
+
+users_to_copy = source_hub.get_users()
+
+if 'totalCount' in users_to_copy and users_to_copy['totalCount'] > 0:
+logging.info("Dumping {} users from {}".format(users_to_copy['totalCount'], source_hub.config['baseurl']))
+assert 'items' in users_to_copy, "Should always be a list of items in a non-zero list of users"
+
+user_data_raw = []
+user_data_prepped_to_copy = []
+
+for user in users_to_copy['items']:
+if user['userName'] == 'sysadmin':
+continue
+# pprint(user)
+user_roles_response = source_hub.get_roles_for_user_or_group(user)
+if user_roles_response:
+user_roles = user_roles_response.json()
+else:
+user_roles = []
+
+user_data_raw.append({
+'user': user, 'roles': user_roles
+})
+
+user_copy_prepped_to_create = user.copy()
+for k in user_keys_to_delete:
+if k in user_copy_prepped_to_create:
+del user_copy_prepped_to_create[k]
+
+# Need to add a password value to get Hub to create the user even
+# for EXTERNAL users
+if 'password' not in user_copy_prepped_to_create:
+user_copy_prepped_to_create['password'] = 'blackduck'
+
+user_roles_to_create = [role['name'] for role in user_roles['items']]
+
+user_data_prepped_to_copy.append({
+'user': user_copy_prepped_to_create, 'roles': user_roles_to_create
+})
+
+raw_file = args.base_filename + ".raw"
+prepped_file = args.base_filename + ".prepped"
+
+serialize_to_file(raw_file, user_data_raw)
+serialize_to_file(prepped_file, user_data_prepped_to_copy)
+else:
+logging.info("No users to dump")
+
+
+