Updating for 1.4 DefectDojo release

Author: aaronweaver

defectdojo_api/__init__.py

@@ -1 +1 @@
-__version__ = '1.1.3'
+__version__ = '1.1.4'

defectdojo_api/__init__.pyc

@@ -80,13 +80,21 @@ def get_test_uri(self, test_id):
         return "/api/" + self.api_version + "/tests/" + str(test_id) + "/"
 
     def get_language_uri(self, language_type_id):
-        """Returns the DefectDojo API URI for a test.
+        """Returns the DefectDojo API URI for a langauge.
 
-        :param test_id: Id of the test
+        :param test_id: Id of the language
 
         """
         return "/api/" + self.api_version + "/language_types/" + str(language_type_id) + "/"
 
+    def get_tool_configuration_uri(self, tool_configuration_id):
+        """Returns the DefectDojo API URI for a tool.
+
+        :param tool_configurations_id: Id of the test
+
+        """
+        return "/api/" + self.api_version + "/tool_configurations/" + str(tool_configuration_id) + "/"
+
     def version_url(self):
         """Returns the DefectDojo API version.
 
@@ -162,8 +170,9 @@ def get_engagement(self, engagement_id):
         return self._request('GET', 'engagements/' + str(engagement_id) + '/')
 
     def create_engagement(self, name, product_id, lead_id, status, target_start, target_end, active='True',
-        pen_test='False', check_list='False', threat_model='False', risk_path="", test_strategy="", progress="",
-        done_testing='False'):
+        pen_test='False', check_list='False', threat_model='False', risk_path="",test_strategy="", progress="",
+        done_testing='False', engagement_type="CI/CD", build_id=None, commit_hash=None, branch_tag=None, build_server=None,
+        source_code_management_server=None, source_code_management_uri=None, orchestration_engine=None, description=None):
         """Creates an engagement with the given properties.
 
         :param name: Engagement name.
@@ -179,6 +188,14 @@ def create_engagement(self, name, product_id, lead_id, status, target_start, tar
         :param risk_path: risk_path
         :param test_strategy: Test Strategy URLs
         :param progress: Engagement progresss measured in percent.
+        :param engagement_type: Interactive or CI/CD
+        :param build_id: Build id from the build server
+        :param commit_hash: Commit hash from source code management
+        :param branch_tag: Branch or tag from source code management
+        :param build_server: Tool Configuration id of build server
+        :param source_code_management_server: URL of source code management
+        :param source_code_management_uri: Link to source code commit
+        :param orchestration_engine: URL of orchestration engine
 
         """
 
@@ -196,9 +213,34 @@ def create_engagement(self, name, product_id, lead_id, status, target_start, tar
             'risk_path': risk_path,
             'test_strategy': test_strategy,
             'progress': progress,
-            'done_testing': done_testing
+            'done_testing': done_testing,
+            'engagement_type': engagement_type
         }
 
+        if description:
+            data.update({'description': description})
+
+        if build_id:
+            data.update({'build_id': build_id})
+
+        if commit_hash:
+            data.update({'commit_hash': commit_hash})
+
+        if branch_tag:
+            data.update({'branch_tag': branch_tag})
+
+        if build_server:
+            data.update({'build_server': self.get_tool_configuration_uri(build_server)})
+
+        if source_code_management_server:
+            data.update({'source_code_management_server': self.get_tool_configuration_uri(source_code_management_server)})
+
+        if source_code_management_uri:
+            data.update({'source_code_management_uri': source_code_management_uri})
+
+        if orchestration_engine:
+            data.update({'orchestration_engine': self.get_tool_configuration_uri(orchestration_engine)})
+
         return self._request('POST', 'engagements/', data=data)
 
     def close_engagement(self, id, user_id=None):
@@ -207,19 +249,13 @@ def close_engagement(self, id, user_id=None):
         :param id: Engagement id.
         :param user_id: User from the user table.
         """
-        engagement = self.get_engagement(id).data
-
-        #if user isn't provided then close with the lead ID
-        if user_id is None:
-            user_id = self.get_id_from_url(engagement["lead"])
 
-        product_id = engagement["product_id"]
-
-        self.set_engagement(id, name=engagement["name"], lead_id=user_id, product_id=product_id, status="Completed", active=False)
+        self.set_engagement(id, status="Completed", active=False)
 
     def set_engagement(self, id, product_id=None, lead_id=None, name=None, status=None, target_start=None,
         target_end=None, active=None, pen_test=None, check_list=None, threat_model=None, risk_path=None,
-        test_strategy=None, progress=None, done_testing=None):
+        test_strategy=None, progress=None, done_testing=None, engagement_type="CI/CD", build_id=None, commit_hash=None, branch_tag=None, build_server=None,
+        source_code_management_server=None, source_code_management_uri=None, orchestration_engine=None, description=None):
 
         """Updates an engagement with the given properties.
 
@@ -237,7 +273,14 @@ def set_engagement(self, id, product_id=None, lead_id=None, name=None, status=No
         :param risk_path: risk_path
         :param test_strategy: Test Strategy URLs
         :param progress: Engagement progresss measured in percent.
-
+        :param engagement_type: Interactive or CI/CD
+        :param build_id: Build id from the build server
+        :param commit_hash: Commit hash from source code management
+        :param branch_tag: Branch or tag from source code management
+        :param build_server: Tool Configuration id of build server
+        :param source_code_management_server: URL of source code management
+        :param source_code_management_uri: Link to source code commit
+        :param orchestration_engine: URL of orchestration engine
         """
 
         data = {}
@@ -284,7 +327,7 @@ def set_engagement(self, id, product_id=None, lead_id=None, name=None, status=No
         if done_testing:
             data['done_testing'] = done_testing
 
-        return self._request('PUT', 'engagements/' + str(id) + '/', data=data)
+        return self._request('PATCH', 'engagements/' + str(id) + '/', data=data)
 
     ###### Product API #######
     def list_products(self, name=None, name_contains=None, limit=20):

defectdojo_api/defectdojo.py

@@ -0,0 +1,86 @@
+"""
+Example written by Aaron Weaver <[email protected]>
+as part of the OWASP DefectDojo and OWASP AppSec Pipeline Security projects
+
+Description: Creates a product in DefectDojo and returns information about the newly created product
+"""
+from defectdojo_api import defectdojo
+
+import os
+import json
+
+# Setup DefectDojo connection information
+host = 'http://localhost:8000'
+api_key = os.environ['DOJO_API_KEY']
+user = 'admin'
+
+"""
+#Optionally, specify a proxy
+proxies = {
+  'http': 'http://localhost:8080',
+  'https': 'http://localhost:8080',
+}
+proxies=proxies
+"""
+
+# Instantiate the DefectDojo api wrapper
+dd = defectdojo.DefectDojoAPI(host, api_key, user, debug=True)
+
+# Add languages to a product
+product_id = 1
+language_type_id = 3
+user_id = 1
+
+dd.delete_all_app_analysis_product(product_id)
+
+#language =
+data = json.load(open('/tmp/wap.json'))
+for app in data["applications"]:
+    name = app["name"]
+    confidence = app["confidence"]
+    version = app["version"]
+    icon = app["icon"]
+    website = app["website"]
+
+    dd.create_app_analysis(product_id, user_id, name, confidence, version, icon, website)
+"""
+#language =
+data = json.load(open('/Users/aweaver/git/AppSecPipelineReports/4cd987e4-6550-48c7-815c-21cf0c4f33fe/reports/cloc/languages.json'))
+
+for language in data:
+    if "header" not in language and "SUM" not in language:
+        print data[language]["code"]
+        files   = data[language]['nFiles']
+        code    = data[language]['code']
+        blank   = data[language]['blank']
+        comment = data[language]['comment']
+        dd.create_language(product_id, user_id, files, code, blank, comment, language_name=language)
+
+#dd.delete_language(1)
+languages = dd.list_language_types(language_name="Python")
+
+if languages.success:
+    for language in languages.data["objects"]:
+        print language['resource_uri']
+"""
+#language_product = dd.list_languages(product_id=1)
+#dd.delete_all_languages_product(1)
+#print language_product
+"""
+# List Products
+products = dd.list_products()
+
+if products.success:
+    #print(products.data_json(pretty=True))  # Decoded JSON object
+    print "********************************"
+    print "Total Number of Products: " + str(products.data["meta"]["total_count"])
+    print "********************************"
+
+    for product in products.data["objects"]:
+        print(product['id'])
+        print(product['name'])  # Print the name of each product
+        print(product['description'])
+        print "******************"
+else:
+    print products.message
+"""

defectdojo_api/defectdojo.pyc

@@ -38,9 +38,6 @@
 print "Creds"
 print list_credentials.data_json(pretty=True)
 
-list_containers = dd.list_containers()
-print "Containers"
-print list_containers.data_json(pretty=True)
 # Search Tool Types by Name
 tool_types = dd.list_tool_types(name="Source Code Repository")