Merge pull request #127 from AR-Calder/managable-monolithic

Split HubRestApi mega-class into multiple files

Author: Glenn Snyder

blackduck/Authentication.py

@@ -0,0 +1,7 @@
+import logging
+import requests
+import json
+from operator import itemgetter
+import urllib.parse
+
+logger = logging.getLogger(__name__)

blackduck/Components.py

@@ -0,0 +1,72 @@
+import logging
+import requests
+import json
+from operator import itemgetter
+import urllib.parse
+
+logger = logging.getLogger(__name__)
+
+def find_component_info_for_protex_component(self, protex_component_id, protex_component_release_id):
+    '''Will return the Hub component corresponding to the protex_component_id, and if a release (version) id
+    is given, the response will also include the component-version. Returns an empty list if there were
+    no components found.
+    '''
+    url = self.config['baseurl'] + "/api/components"
+    if protex_component_release_id:
+        query = "?q=bdsuite:{}%23{}&limit=9999".format(protex_component_id, protex_component_release_id)
+    else:
+        query = "?q=bdsuite:{}&limit=9999".format(protex_component_id)
+    with_query = url + query
+    logger.debug("Finding the Hub componet for Protex component id {}, release id {} using query/url {}".format(
+        protex_component_id, protex_component_release_id, with_query))
+    response = self.execute_get(with_query)
+    logger.debug("query results in status code {}, json data: {}".format(response.status_code, response.json()))
+    # TODO: Error checking and retry? For now, as POC just assuming it worked
+    component_list_d = response.json()
+    return response.json()
+
+def _get_components_url(self):
+    return self.get_urlbase() + "/api/components"
+
+def get_components(self, limit=100, parameters={}):
+    if limit:
+        parameters.update({'limit':limit})
+    #
+    # I was only able to GET components when using this internal media type which is how the GUI works
+    #       July 19, 2019 Glenn Snyder
+    #
+    custom_headers = {'Accept':'application/vnd.blackducksoftware.internal-1+json'}
+    url = self._get_components_url() + self._get_parameter_string(parameters)
+    response = self.execute_get(url, custom_headers=custom_headers)
+    return response.json()
+
+def search_components(self, search_str_or_query, limit=100, parameters={}):
+    if limit:
+        parameters.update({'limit':limit})
+    if search_str_or_query.startswith("q="):
+        # allow caller to override original behavior with their own query
+        query = search_str_or_query
+    else:
+        # maintain original, somewhat flawed behavior
+        query = "q=name:{}".format(search_str_or_query)
+    parm_str = self._get_parameter_string(parameters)
+    url = self.get_apibase() + "/search/components{}&{}".format(parm_str, query)
+    response = self.execute_get(url)
+    return response.json()
+    
+def get_component_by_id(self, component_id):
+    url = self.config['baseurl'] + "/api/components/{}".format(component_id)
+    return self.get_component_by_url(url)
+
+def get_component_by_url(self, component_url):
+    headers = self.get_headers()
+    response = self.execute_get(component_url)
+    jsondata = response.json()
+    return jsondata
+
+def update_component_by_id(self, component_id, update_json):
+    url = self.config["baseurl"] + "/api/components/{}".format(component_id)
+    return self.update_component_by_url(url, update_json)
+
+def update_component_by_url(self, component_url, update_json):
+    return self.execute_put(component_url, update_json)

blackduck/Core.py

@@ -0,0 +1,183 @@
+import logging
+import requests
+import json
+from operator import itemgetter
+import urllib.parse
+
+logger = logging.getLogger(__name__)
+
+def read_config(self):
+    try:
+        with open('.restconfig.json','r') as f:
+            self.config = json.load(f)
+    except:
+        logging.error(f"Unable to load configuration from '.restconfig.json'. Make sure you create one with proper connection and authentication values for your Black Duck server")
+        raise
+        
+def write_config(self):
+    with open(self.configfile,'w') as f:
+        json.dump(self.config, f, indent=3)
+        
+def get_auth_token(self):
+    api_token = self.config.get('api_token', False)
+    if api_token:
+        authendpoint = "/api/tokens/authenticate"
+        url = self.config['baseurl'] + authendpoint
+        session = requests.session()
+        response = session.post(
+            url, 
+            data={}, 
+            headers={'Authorization': 'token {}'.format(api_token)}, 
+            verify=not self.config['insecure']
+        )
+        csrf_token = response.headers['X-CSRF-TOKEN']
+        try:
+            bearer_token = json.loads(response.content.decode('utf-8'))['bearerToken']
+        except json.decoder.JSONDecodeError as e:
+            logger.exception("Authentication failure, could not obtain bearer token")
+            raise Exception("Failed to obtain bearer token, check for valid authentication token")
+        return (bearer_token, csrf_token, None)
+    else:
+        authendpoint="/j_spring_security_check"
+        url = self.config['baseurl'] + authendpoint
+        session=requests.session()
+        credentials = dict()
+        credentials['j_username'] = self.config['username']
+        credentials['j_password'] = self.config['password']
+        response = session.post(url, credentials, verify= not self.config['insecure'])
+        cookie = response.headers['Set-Cookie']
+        token = cookie[cookie.index('=')+1:cookie.index(';')]
+    return (token, None, cookie)
+
+def _get_hub_rest_api_version_info(self):
+    '''Get the version info from the server, if available
+    '''
+    session = requests.session()
+    url = self.config['baseurl'] + "/api/current-version"
+    response = session.get(url, verify = not self.config['insecure'])
+
+    if response.status_code == 200:
+        version_info = response.json()
+        if 'version' in version_info:
+            return version_info
+        else:
+            raise UnknownVersion("Did not find the 'version' key in the response to a successful GET on /api/current-version")
+    else:
+        raise UnknownVersion("Failed to retrieve the version info from {}, status code {}".format(url, response.status_code))
+
+def _get_major_version(self):
+    return self.version_info['version'].split(".")[0]
+
+def get_urlbase(self):
+    return self.config['baseurl']
+
+def get_headers(self):
+    if self.config.get('api_token', False):
+        return {
+            'X-CSRF-TOKEN': self.csrf_token, 
+            'Authorization': 'Bearer {}'.format(self.token), 
+            'Accept': 'application/json',
+            'Content-Type': 'application/json'}
+    else:
+        if self.bd_major_version == "3":
+            return {"Cookie": self.cookie}
+        else:
+            return {"Authorization":"Bearer " + self.token}
+
+def get_api_version(self):
+    url = self.get_urlbase() + '/api/current-version'
+    response = self.execute_get(url)
+    version = response.json().get('version', 'unknown')
+    return version
+
+def _get_parameter_string(self, parameters={}):
+    parameter_string = "&".join(["{}={}".format(k,urllib.parse.quote(str(v))) for k,v in sorted(parameters.items(), key=itemgetter(0))])
+    return "?" + parameter_string
+
+def get_tags_url(self, component_or_project):
+    # Utility method to return the tags URL from either a component or project object
+    url = None
+    for link_d in component_or_project['_meta']['links']:
+        if link_d['rel'] == 'tags':
+            return link_d['href']
+    return url
+
+def get_link(self, bd_rest_obj, link_name):
+    # returns the URL for the link_name OR None
+    if bd_rest_obj and '_meta' in bd_rest_obj and 'links' in bd_rest_obj['_meta']:
+        for link_obj in bd_rest_obj['_meta']['links']:
+            if 'rel' in link_obj and link_obj['rel'] == link_name:
+                return link_obj.get('href', None)
+    else:
+        logger.warning("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"
+
+def execute_delete(self, url):
+    headers = self.get_headers()
+    response = requests.delete(url, headers=headers, verify = not self.config['insecure'])
+    return response
+
+def _validated_json_data(self, data_to_validate):
+    if isinstance(data_to_validate, dict) or isinstance(data_to_validate, list):
+        json_data = json.dumps(data_to_validate)
+    else:
+        json_data = data_to_validate
+    json.loads(json_data) # will fail with JSONDecodeError if invalid
+    return json_data
+
+def execute_get(self, url, custom_headers={}):
+    headers = self.get_headers()
+    headers.update(custom_headers)
+    response = requests.get(url, headers=headers, verify = not self.config['insecure'])
+    return response
+    
+def execute_put(self, url, data, custom_headers={}):
+    json_data = self._validated_json_data(data)
+    headers = self.get_headers()
+    headers["Content-Type"] = "application/json"
+    headers.update(custom_headers)
+    response = requests.put(url, headers=headers, data=json_data, verify = not self.config['insecure'])
+    return response
+
+def _create(self, url, json_body):
+    response = self.execute_post(url, json_body)
+    # 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"]
+        else:
+            try:
+                response_json = response.json()
+            except json.decoder.JSONDecodeError:
+                logger.warning('did not receive any json data back')
+            else:
+                if '_meta' in response_json and 'href' in response_json['_meta']:
+                    return response_json['_meta']['href']
+                else:
+                    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:
+        raise CreateFailedUnknown("Failed to create the object for an unknown reason - url {}, body {}, response {}".format(url, json_body, response))
+
+def execute_post(self, url, data, custom_headers={}):
+    json_data = self._validated_json_data(data)
+    headers = self.get_headers()
+    headers["Content-Type"] = "application/json"
+    headers.update(custom_headers)
+    response = requests.post(url, headers=headers, data=json_data, verify = not self.config['insecure'])
+    return response
+
+def get_matched_components(self, version_obj, limit=9999):
+    url = "{}/matched-files".format(version_obj['_meta']['href'])
+    param_string = self._get_parameter_string({'limit': limit})
+    url = "{}{}".format(url, param_string)
+    response = self.execute_get(url)
+    return response.json()

blackduck/CustomFields.py

@@ -0,0 +1,112 @@
+import logging
+import requests
+import json
+from operator import itemgetter
+import urllib.parse
+
+logger = logging.getLogger(__name__)
+
+def _get_cf_url(self):
+    return self.get_apibase() + "/custom-fields/objects"
+
+def supported_cf_object_types(self):
+    '''Get the types and cache them since they are static (on a per-release basis)'''
+    if not hasattr(self, "_cf_object_types"):
+        logger.debug("retrieving object types")
+        self._cf_object_types = [cfo['name'] for cfo in self.get_cf_objects().get('items', [])]
+    return self._cf_object_types
+
+def get_cf_objects(self):
+    '''Get CF objects and cache them since these are static (on a per-release basis)'''
+    url = self._get_cf_url()
+    if not hasattr(self, "_cf_objects"):
+        logger.debug("retrieving objects")
+        response = self.execute_get(url)
+        self._cf_objects = response.json()
+    return self._cf_objects
+
+def _get_cf_object_url(self, object_name):
+    for cf_object in self.get_cf_objects().get('items', []):
+        if cf_object['name'].lower() == object_name.lower():
+            return cf_object['_meta']['href']
+
+def get_cf_object(self, object_name):
+    assert object_name in self.supported_cf_object_types(), "Object name {} not one of the supported types ({})".format(object_name, self.supported_cf_object_types())
+
+    object_url = self._get_cf_object_url(object_name)
+    response = self.execute_get(object_url)
+    return response.json()
+
+def _get_cf_obj_rel_path(self, object_name):
+    return object_name.lower().replace(" ", "-")
+
+def create_cf(self, object_name, field_type, description, label, position, active=True, initial_options=[]):
+    '''
+        Create a custom field for the given object type (e.g. "Project", "Project Version") using the field_type and other parameters.
+
+        Initial options are needed for field types like multi-select where the multiple values to choose from must also be provided.
+
+        initial_options = [{"label":"val1", "position":0}, {"label":"val2", "position":1}]
+    '''
+    assert isinstance(position, int) and position >= 0, "position must be an integer that is greater than or equal to 0"
+    assert field_type in ["BOOLEAN", "DATE", "DROPDOWN", "MULTISELECT", "RADIO", "TEXT", "TEXTAREA"]
+
+    types_using_initial_options = ["DROPDOWN", "MULTISELECT", "RADIO"]
+
+    post_url = self._get_cf_object_url(object_name) + "/fields"
+    cf_object = self._get_cf_obj_rel_path(object_name)
+    cf_request = {
+        "active": active,
+        "description": description,
+        "label": label,
+        "position": position,
+        "type": field_type,
+    }
+    if field_type in types_using_initial_options and initial_options:
+        cf_request.update({"initialOptions": initial_options})
+    response = self.execute_post(post_url, data=cf_request)
+    return response
+
+def delete_cf(self, object_name, field_id):
+    '''Delete a custom field from a given object type, e.g. Project, Project Version, Component, etc
+
+    WARNING: Deleting a custom field is irreversiable. Any data in the custom fields could be lost so use with caution.
+    '''
+    assert object_name in self.supported_cf_object_types(), "You must supply a supported object name that is in {}".format(self.supported_cf_object_types())
+
+    delete_url = self._get_cf_object_url(object_name) + "/fields/{}".format(field_id)
+    return self.execute_delete(delete_url)
+
+def get_custom_fields(self, object_name):
+    '''Get the custom field (definition) for a given object type, e.g. Project, Project Version, Component, etc
+    '''
+    assert object_name in self.supported_cf_object_types(), "You must supply a supported object name that is in {}".format(self.supported_cf_object_types())
+
+    url = self._get_cf_object_url(object_name) + "/fields"
+
+    response = self.execute_get(url)        
+    return response.json()
+
+def get_cf_values(self, obj):
+    '''Get all of the custom fields from an object such as a Project, Project Version, Component, etc
+
+    The obj is expected to be the JSON document for a project, project-version, component, etc
+    '''
+    url = self.get_link(obj, "custom-fields")
+    response = self.execute_get(url)
+    return response.json()
+
+def get_cf_value(self, obj, field_id):
+    '''Get a custom field value from an object such as a Project, Project Version, Component, etc
+
+    The obj is expected to be the JSON document for a project, project-version, component, etc
+    '''
+    url = self.get_link(obj, "custom-fields") + "/{}".format(field_id)
+    response = self.execute_get(url)
+    return response.json()
+
+def put_cf_value(self, cf_url, new_cf_obj):
+    '''new_cf_obj is expected to be a modified custom field value object with the values updated accordingly, e.g.
+    call get_cf_value, modify the object, and then call put_cf_value
+    '''
+    return self.execute_put(cf_url, new_cf_obj)