code styling
Some checks failed
ci/woodpecker/push/woodpecker Pipeline failed

This commit is contained in:
2025-07-30 09:08:33 +02:00
parent 98035fe59c
commit 955937dea7
3 changed files with 249 additions and 137 deletions

View File

@@ -1,5 +1,6 @@
import os
import sys
import subprocess
from loguru import logger
import json
@@ -7,8 +8,8 @@ import json
import datetime
from dateutil.relativedelta import relativedelta
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'defectdojo-client'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), 'dependencytrack-client'))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "defectdojo-client"))
sys.path.insert(0, os.path.join(os.path.dirname(__file__), "dependencytrack-client"))
import defectdojo_api
from defectdojo_api.rest import ApiException as DefectDojoApiException
@@ -16,6 +17,7 @@ from defectdojo_api.rest import ApiException as DefectDojoApiException
import dependencytrack_api
from dependencytrack_api.rest import ApiException as DependencyTrackApiException
class ApiException(Exception):
def __init__(self, cause):
self.cause = cause
@@ -25,11 +27,14 @@ class ApiException(Exception):
self.data = cause.data
self.headers = cause.headers
class ApiCallExecutor:
def __init__(self, verbose):
self.verbose = verbose
def innerExecuteApiCall(self, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams):
def innerExecuteApiCall(
self, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams
):
logger.info(f"Calling {ApiClass=}.{EndpointMethod=} with {RequestClass=})")
if self.verbose:
logger.debug(f"{additionalParams=}, {requestParams=}")
@@ -42,38 +47,57 @@ class ApiCallExecutor:
logger.info(f"Response is {response}")
return response
class DefectDojoApiClient(defectdojo_api.ApiClient, ApiCallExecutor):
def __init__(self, config, verbose):
defectdojo_api.ApiClient.__init__(self, config)
ApiCallExecutor.__init__(self, verbose)
def executeApiCall(self, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams):
def executeApiCall(
self, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams
):
try:
return self.innerExecuteApiCall(ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams)
return self.innerExecuteApiCall(
ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams
)
except defectdojo_api.exceptions.ApiException as e:
raise ApiException(e)
class DependencyTrackApiClient(dependencytrack_api.ApiClient, ApiCallExecutor):
def __init__(self, config, verbose):
dependencytrack_api.ApiClient.__init__(self, config)
ApiCallExecutor.__init__(self, verbose)
def executeApiCall(self, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams):
def executeApiCall(
self, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams
):
try:
return self.innerExecuteApiCall(ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams)
return self.innerExecuteApiCall(
ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams
)
except dependencytrack_api.exceptions.ApiException as e:
raise ApiException(e)
def generateSBOM(target='.', name='dummyName', version='0.0.0'):
def generateSBOM(target=".", name="dummyName", version="0.0.0"):
try:
result = subprocess.run(
["syft", "scan", target, "-o", "cyclonedx-json", "--source-name", name, "--source-version", version],
[
"syft",
"scan",
target,
"-o",
"cyclonedx-json@1.5",
"--source-name",
name,
"--source-version",
version,
],
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True
text=True,
)
sbom = json.loads(result.stdout)
return sbom
@@ -82,28 +106,39 @@ def generateSBOM(target='.', name='dummyName', version='0.0.0'):
raise MyLocalException(e)
def loadToDTrackAndDefectDojo(config, projectName, projectVersion, projectClassifier, projectDescription, productType, sbom, reImport):
def loadToDTrackAndDefectDojo(
config,
projectName,
projectVersion,
projectClassifier,
projectDescription,
productType,
sbom,
reImport,
):
# ------- create product and engagement in DefectDojo -------
if not reImport:
# in case of a reimport no modification on DefectDojo are required
defectdojo_configuration = defectdojo_api.Configuration(
host = config['DEFECTDOJO_URL']
host=config["DEFECTDOJO_URL"]
)
defectdojo_configuration.api_key['tokenAuth'] = config['DEFECTDOJO_TOKEN']
defectdojo_configuration.api_key_prefix['tokenAuth'] = 'Token'
defectdojo_configuration.api_key["tokenAuth"] = config["DEFECTDOJO_TOKEN"]
defectdojo_configuration.api_key_prefix["tokenAuth"] = "Token"
with DefectDojoApiClient(defectdojo_configuration, config['VERBOSE']) as client:
with DefectDojoApiClient(defectdojo_configuration, config["VERBOSE"]) as client:
print("Create product in DefectDojo")
productName = f"{projectName}:{projectVersion}"
product_response = \
client.executeApiCall(
defectdojo_api.ProductsApi,
defectdojo_api.ProductsApi.products_create,
defectdojo_api.ProductRequest,
{ 'name': productName, 'description': projectDescription, 'prod_type': productType },
[]
)
product_response = client.executeApiCall(
defectdojo_api.ProductsApi,
defectdojo_api.ProductsApi.products_create,
defectdojo_api.ProductRequest,
{
"name": productName,
"description": projectDescription,
"prod_type": productType,
},
[],
)
product_id = product_response.id
print(f"{product_id=}")
@@ -112,48 +147,71 @@ def loadToDTrackAndDefectDojo(config, projectName, projectVersion, projectClassi
start_time = datetime.date.today()
end_time = start_time + relativedelta(years=10)
engagementName = f"{productName} DTrack Link"
engagement_response = \
client.executeApiCall(
defectdojo_api.EngagementsApi,
defectdojo_api.EngagementsApi.engagements_create,
defectdojo_api.EngagementRequest,
{ 'name': engagementName, 'target_start': start_time, 'target_end': end_time, 'status': 'In Progress', 'product': product_id },
[]
)
engagement_response = client.executeApiCall(
defectdojo_api.EngagementsApi,
defectdojo_api.EngagementsApi.engagements_create,
defectdojo_api.EngagementRequest,
{
"name": engagementName,
"target_start": start_time,
"target_end": end_time,
"status": "In Progress",
"product": product_id,
},
[],
)
engagement_id = engagement_response.id
print(f"{engagement_id=}")
# ------- create project in DependencyTrack, connect project to engagement in DefectDojo, upload SBOM --------
# ------- create project in DependencyTrack, connect project to engagement in DefectDojo, upload SBOM --------
dependencytrack_configuration = dependencytrack_api.Configuration(
host = f"{config['DTRACK_API_URL']}/api"
host=f"{config['DTRACK_API_URL']}/api"
)
dependencytrack_configuration.debug = False
dependencytrack_configuration.api_key['ApiKeyAuth'] = config['DTRACK_TOKEN']
dependencytrack_configuration.api_key["ApiKeyAuth"] = config["DTRACK_TOKEN"]
with DependencyTrackApiClient(dependencytrack_configuration, config['VERBOSE']) as client:
with DependencyTrackApiClient(
dependencytrack_configuration, config["VERBOSE"]
) as client:
if not reImport:
# in case of a reimport it is not necessary to create the project
project_response = \
client.executeApiCall(
dependencytrack_api.ProjectApi,
dependencytrack_api.ProjectApi.create_project,
dependencytrack_api.Project,
{ 'name': projectName, 'version': projectVersion, 'classifier': projectClassifier, 'uuid': "", 'last_bom_import': 0 },
[]
)
project_response = client.executeApiCall(
dependencytrack_api.ProjectApi,
dependencytrack_api.ProjectApi.create_project,
dependencytrack_api.Project,
{
"name": projectName,
"version": projectVersion,
"classifier": projectClassifier,
"uuid": "",
"last_bom_import": 0,
},
[],
)
project_uuid = project_response.uuid
print(f"{project_uuid=}")
properties = [
{ 'group_name': "integrations", 'property_name': "defectdojo.engagementId",
'property_value': str(engagement_id), 'property_type': "STRING" },
{ 'group_name': "integrations", 'property_name': "defectdojo.doNotReactivate",
'property_value': "true", 'property_type': "BOOLEAN" },
{ 'group_name': "integrations", 'property_name': "defectdojo.reimport",
'property_value': "true", 'property_type': "BOOLEAN" }
{
"group_name": "integrations",
"property_name": "defectdojo.engagementId",
"property_value": str(engagement_id),
"property_type": "STRING",
},
{
"group_name": "integrations",
"property_name": "defectdojo.doNotReactivate",
"property_value": "true",
"property_type": "BOOLEAN",
},
{
"group_name": "integrations",
"property_name": "defectdojo.reimport",
"property_value": "true",
"property_type": "BOOLEAN",
},
]
for property in properties:
client.executeApiCall(
@@ -161,15 +219,24 @@ def loadToDTrackAndDefectDojo(config, projectName, projectVersion, projectClassi
dependencytrack_api.ProjectPropertyApi.create_property1,
dependencytrack_api.ProjectProperty,
property,
[ project_uuid ]
[project_uuid],
)
bom_response = \
client.executeApiCall(
dependencytrack_api.BomApi,
dependencytrack_api.BomApi.upload_bom,
bom_response = client.executeApiCall(
dependencytrack_api.BomApi,
dependencytrack_api.BomApi.upload_bom,
None,
None,
[
None,
False,
projectName,
projectVersion,
None,
None,
[ None, False, projectName, projectVersion, None, None, None, None, True, sbom ]
)
None,
None,
True,
sbom,
],
)