dtrack-defectdojo-automation/snippets/test03.py

import os
from loguru import logger
import argparse
import subprocess
import json

import defectdojo_api
from defectdojo_api.rest import ApiException as DefectDojoApiException
import datetime
from dateutil.relativedelta import relativedelta

import dependencytrack_api
from dependencytrack_api.rest import ApiException as DependencyTrackApiException

class MyLocalException(Exception): pass

def executeApiCall(apiClient, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams=[]):
    try:
        logger.info(f"Calling {ApiClass}.{EndpointMethod} with {RequestClass} ({additionalParams}, {requestParams})")
        instance = ApiClass(apiClient)
        if RequestClass:
            request = RequestClass(**requestParams)
            response = EndpointMethod(instance, *additionalParams, request)
        else:
            response = EndpointMethod(instance, *additionalParams)
        logger.info(f"Response is {response}")
        return response
    except Exception as e:
        logger.error(f"Caught error {e} with {str(e)}")
        raise MyLocalException(e)

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],
            check=True,
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True
        )
        sbom = json.loads(result.stdout)
        return sbom
    except subprocess.CalledProcessError as e:
        logger.error(f"SBOM scanner failed: {e.stderr}")
        raise MyLocalException(e)


try:
    DTRACK_API_URL = os.environ["DTRACK_API_URL"]
    DTRACK_TOKEN = os.environ["DTRACK_TOKEN"]
    DEFECTDOJO_URL = os.environ["DEFECTDOJO_URL"]
    DEFECTDOJO_TOKEN = os.environ["DEFECTDOJO_TOKEN"]
except KeyError as e:
    raise Exception(f"Env variable {e} is shall be set")


parser = argparse.ArgumentParser(description='snippet/test03')
parser.add_argument('--name', '-n',
                    help='Project Name',
                    required=True)
parser.add_argument('--version', '-v',
                    help='Project Version',
                    required=True)
parser.add_argument('--description', '-d',
                    help='Project Description',
                    required=True)
parser.add_argument('--type', '-t',
                    help='Product Type from DefectDojo',
                    type=int,
                    required=True)
parser.add_argument('--classifier', '-c',
                    help='Project Classifier from DependencyTrack',
                    choices=['APPLICATION', 'FRAMEWORK', 'LIBRARY', 'CONTAINER', 'OPERATING_SYSTEM', 'DEVICE', 'FIRMWARE', 'FILE', 'PLATFORM', 'DEVICE_DRIVER', 'MACHINE_LEARNING_MODEL', 'DATA'],
                    required=True)
parser.add_argument('--target', '-T',
                    help='Target to scan, either path name for sources or docker image tag',
                    required=True)
args = parser.parse_args()
projectName = args.name
projectVersion = args.version
projectDescription = args.description
productType = args.type
projectClassifier = args.classifier
target = args.target


logger.info(f"Generating SBOM for {target}")
sbom = generateSBOM(target, projectName, projectVersion)
logger.info("Done.")


defectdojo_configuration = defectdojo_api.Configuration(
    host = DEFECTDOJO_URL
)
defectdojo_configuration.api_key['tokenAuth'] = DEFECTDOJO_TOKEN
defectdojo_configuration.api_key_prefix['tokenAuth'] = 'Token'

dependencytrack_configuration = dependencytrack_api.Configuration(
    host = f"{DTRACK_API_URL}/api"
)
dependencytrack_configuration.debug = False
dependencytrack_configuration.api_key['ApiKeyAuth'] = DTRACK_TOKEN

with defectdojo_api.ApiClient(defectdojo_configuration) as defectdojo_api_client:
    print("Create product in DefectDojo")
    productName = f"{projectName}:{projectVersion}"
    product_response = \
        executeApiCall(
           defectdojo_api_client, 
           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=}")

    print("Create engagement in DefectDojo")
    start_time = datetime.date.today()
    end_time = start_time + relativedelta(years=10)
    engagementName = f"{productName} DTrack Link"
    engagement_response = \
        executeApiCall(
           defectdojo_api_client, 
           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=}")

with dependencytrack_api.ApiClient(dependencytrack_configuration) as dependencytrack_api_client:
    project_response = \
        executeApiCall(
           dependencytrack_api_client,
           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" }
    ]
    for property in properties:
        executeApiCall(
            dependencytrack_api_client,
            dependencytrack_api.ProjectPropertyApi,
            dependencytrack_api.ProjectPropertyApi.create_property1,
            dependencytrack_api.ProjectProperty,
            property,
            [ project_uuid ]
        )

    bom_response = \
        executeApiCall(
            dependencytrack_api_client,
            dependencytrack_api.BomApi,
            dependencytrack_api.BomApi.upload_bom,
            None,
            None,
            [ None, False, projectName, projectVersion, None, None, None, None, True, json.dumps(sbom) ]
        )