works

.gitlab-ci.yml

@@ -1,6 +1,6 @@
 stages:
   - generate-api-clients
-  - dockerize
+  - build
 
 variables:
   REGISTRY: devnexus.krohne.com:18079/repository/docker-krohne
@@ -69,7 +69,7 @@ generate-defectdojo-api:
         --package-name defectdojo_api 
 
 dockerize:
-  stage: dockerize
+  stage: build
   image: devnexus.krohne.com:18079/repository/docker-krohne/krohnedockerbash:0.5
   tags:
     - linux
@@ -89,38 +89,38 @@ dockerize:
     - docker push $IMAGE_NAME:$CI_COMMIT_SHA
     - docker push $IMAGE_NAME:$CI_COMMIT_TAG
 
+build-windows-binary:
+  stage: build
+  tags:
+    - windows
+    - pwsh
+    - python3.13
+  rules:
+    - if: '$CI_COMMIT_TAG'
+  artifacts:
+    paths:
+      - sbom-dt-dd.exe
+  script:
+    - |
+      cd src
+      mv ..\dependencytrack-client .
+      mv ..\defectdojo-client .
+      & 'C:\Program Files\Python313\python.exe' -m venv venv
+      .\venv\Scripts\pip.exe install --upgrade pip
+      .\venv\Scripts\pip.exe install -r requirements.txt
+      .\venv\Scripts\pip.exe install -r dependencytrack-client\requirements.txt
+      .\venv\Scripts\pip.exe install -r defectdojo-client\requirements.txt
+      .\venv\Scripts\pip.exe install pyinstaller
+      .\venv\Scripts\pyinstaller.exe --onefile `
+        --add-data "dependencytrack-client;dependencytrack-client" `
+        --add-data "defectdojo-client;defectdojo-client" `
+        --hidden-import pydantic `
+        --hidden-import dateutil.parser `
+        --hidden-import urllib3 `
+        --hidden-import regex `
+        --collect-data cyclonedx `
+        --collect-data license_experssion `
+        sbom-dt-dd.py
+      mv dist\sbom-dt-dd.exe ..
 
 
-#
-#  build:
-#    image: plugins/kaniko
-#    settings:
-#      repo: ${FORGE_NAME}/${CI_REPO}
-#      registry:
-#        from_secret: container_registry
-#      tags: latest,${CI_COMMIT_SHA},${CI_COMMIT_TAG}
-#      username:
-#        from_secret: container_registry_username
-#      password:
-#        from_secret: container_registry_password
-#      dockerfile: Dockerfile
-#    when:
-#      - event: [ push, tag ]
-#
-#  build-for-quay:
-#    image: plugins/kaniko
-#    settings:
-#      repo: quay.io/wollud1969/${CI_REPO_NAME}
-#      registry: quay.io
-#      tags: 
-#        - latest
-#        - ${CI_COMMIT_TAG}
-#      username:
-#        from_secret: quay_username
-#      password:
-#        from_secret: quay_password
-#      dockerfile: Dockerfile
-#    when:
-#      - event: [tag]
-#
-

Dockerfile

@@ -31,7 +31,8 @@ USER user
 WORKDIR $APP_DIR
 
 COPY src/requirements.txt .
-COPY src/sbom-dt-dd.py .
+COPY src/sbom_dt_dd.py .
+COPY src/sbom_dt_dd_cli.py .
 COPY src/converter.py .
 COPY src/entrypoint.sh .
 COPY dependencytrack-client/ ./dependencytrack-client

snippets/websrv/main.py

@@ -0,0 +1,33 @@
+from fastapi import FastAPI
+from fastapi.responses import JSONResponse
+
+app = FastAPI(
+    title="My FastAPI App",
+    version="1.0.0",
+    description="A simple FastAPI example with uvicorn and gunicorn."
+)
+
+@app.get("/hello")
+async def say_hello(name: str):
+    """
+    Returns a friendly greeting.
+    ---
+    parameters:
+      - name: name
+        in: query
+        required: true
+        schema:
+          type: string
+    responses:
+      200:
+        description: Successful Response
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                message:
+                  type: string
+    """
+    return JSONResponse(content={"message": f"Hello, {name}!"})
+

snippets/websrv/requirements.txt

@@ -0,0 +1,3 @@
+fastapi==0.116.1
+gunicorn==23.0.0
+uvicorn==0.35.0

snippets/websrv/server.sh

@@ -0,0 +1,4 @@
+#!/bin/bash
+
+./.venv/bin/gunicorn main:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000
+

src/converter.py

@@ -47,7 +47,7 @@ def __converterClassifierToComponentType(classifier):
 
 
 
-def minimalSbomFormatConverter(minimalSbom, classifier):
+def minimalSbomFormatConverter(minimalSbom):
     logger.info(f"Minimal input: {minimalSbom}")
 
     lc_factory = LicenseFactory()

src/entrypoint.sh

@@ -7,5 +7,4 @@ PYTHONPATH="$PYTHONPATH:/opt/app/dependencytrack-client"
 PYTHONPATH="$PYTHONPATH:/opt/app/defectdojo-client"
 export PYTHONPATH
 
-exec python /opt/app/sbom-dt-dd.py "$@"
+exec python /opt/app/sbom_dt_dd_cli.py "$@"
-

src/requirements.txt

@@ -2,4 +2,6 @@ regex==2024.11.6
 loguru==0.7.3
 PyYAML==6.0.2
 cyclonedx-python-lib==10.4.1
-
+fastapi==0.116.1
+gunicorn==23.0.0
+uvicorn==0.35.0

src/sbom-dt-dd.py

@@ -1,259 +0,0 @@
-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
-
-from converter import minimalSbomFormatConverter
-
-
-class MyLocalException(Exception): pass
-
-def executeApiCall(apiClient, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams=[]):
-    try:
-        logger.info(f"Calling {ApiClass=}.{EndpointMethod=} with {RequestClass=})")
-        if VERBOSE:
-            logger.debug(f"{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)
-
-# ---- main starts here with preparation of config -----------------------------------------------------------------------
-
-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='sbom-dt-dd glue logic')
-parser.add_argument('--name', '-n',
-                    help='Project Name',
-                    required=False,
-                    default=''),
-parser.add_argument('--version', '-v',
-                    help='Project Version',
-                    required=False,
-                    default='')
-parser.add_argument('--description', '-d',
-                    help='Project Description',
-                    required=False,
-                    default='')
-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=False,
-                    default='')
-parser.add_argument('--uploadsbom', '-U',
-                    help='Upload a already existing SBOM instead of generating it. Give the SBOM file at -F instead of a target',
-                    required=False,
-                    action='store_true',
-                    default=False)
-parser.add_argument('--sbomfile', '-F',
-                    help='Filename of existing SBOM file to upload, use together with -U, do not use together with -T',
-                    required=False)
-parser.add_argument('--minimalsbomformat', '-K',
-                    help='SBOM file comes in dedicated minimal format and will be converted into cyclonedx before uploading',
-                    action='store_true',
-                    default=False)
-parser.add_argument('--overwritemetadata', '-O',
-                    help='Overwrite name, version, description and classifier with data from minimal SBOM',
-                    action='store_true',
-                    default=False)
-parser.add_argument('--target', '-T',
-                    help='Target to scan, either path name for sources or docker image tag',
-                    required=False)
-parser.add_argument('--reimport', '-R',
-                    help='Import the SBOM for an existing project/product once again',
-                    required=False,
-                    action='store_true',
-                    default=False)
-parser.add_argument('--verbose', '-V',
-                    help='A lot of debug output',
-                    required=False,
-                    action='store_true',
-                    default=False)
-args = parser.parse_args()
-projectName = args.name
-projectVersion = args.version
-projectDescription = args.description
-productType = args.type
-projectClassifier = args.classifier
-reImport = args.reimport
-
-uploadSbomFlag = args.uploadsbom
-if uploadSbomFlag:
-    sbomFileName = args.sbomfile
-    minimalSbomFormat = args.minimalsbomformat
-else:
-    target = args.target
-
-if minimalSbomFormat:
-    overwriteMetadata = args.overwritemetadata
-
-if not overwriteMetadata and not (projectName and projectVersion and projectClassifier and projectDescription):
-    raise MyLocalException("If overwriteMetadata is not selected, projectName, projectVersion, projectClassifier and projectDescription must be set.")
-
-VERBOSE = args.verbose
-
-
-# ---- main starts here --------------------------------------------------------------------------------------------------
-
-if uploadSbomFlag:
-    # ------- read uploaded SBOM -------------
-    logger.info(f"Reading SBOM from file {sbomFileName}")
-    with open(sbomFileName, 'r') as sbomFile:
-        sbom = sbomFile.read()
-    logger.info("SBOM file read.")
-    if minimalSbomFormat:
-        logger.info("Start converting from minimal format into cyclonedx")
-        (sbom, nameFromMinimalSbom, versionFromMinimalSbom, classifierFromMinimalSbom, descriptionFromMinimalSbom) = minimalSbomFormatConverter(sbom, projectClassifier)
-        logger.info("Converted")
-        if overwriteMetadata:
-            projectName = nameFromMinimalSbom
-            projectVersion = versionFromMinimalSbom
-            projectClassifier = classifierFromMinimalSbom
-            projectDescription = descriptionFromMinimalSbom
-    logger.info("Done.")
-else:
-    # ------- generate SBOM ------------
-    logger.info(f"Generating SBOM for {target}")
-    sbomJson = generateSBOM(target, projectName, projectVersion)
-    sbom = json.dumps(sbomJson)
-    logger.info("Done.")
-
-
-
-# ------- 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 = DEFECTDOJO_URL
-    )
-    defectdojo_configuration.api_key['tokenAuth'] = DEFECTDOJO_TOKEN
-    defectdojo_configuration.api_key_prefix['tokenAuth'] = '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=}")
-
-
-# ------- create project in DependencyTrack, connect project to engagement in DefectDojo, upload SBOM --------
-dependencytrack_configuration = dependencytrack_api.Configuration(
-    host = f"{DTRACK_API_URL}/api"
-)
-dependencytrack_configuration.debug = False
-dependencytrack_configuration.api_key['ApiKeyAuth'] = DTRACK_TOKEN
-
-with dependencytrack_api.ApiClient(dependencytrack_configuration) as dependencytrack_api_client:
-    if not reImport:
-        # in case of a reimport it is not necessary to create the project
-        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, sbom ]
-        )
-

src/sbom_dt_dd.py

@@ -0,0 +1,174 @@
+import os
+import sys
+
+from loguru import logger
+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'))
+
+import defectdojo_api
+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, status, reason, body, data, headers):
+        self.status = status
+        self.reason = reason
+        self.body = body
+        self.data = data
+        self.headers = None
+
+class ApiCallExecutor:
+    def __init__(self, verbose):
+        self.verbose = verbose
+
+    def innerExecuteApiCall(self, ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams):
+        logger.info(f"Calling {ApiClass=}.{EndpointMethod=} with {RequestClass=})")
+        if self.verbose:
+            logger.debug(f"{additionalParams=}, {requestParams=}")
+        instance = ApiClass(self)
+        if RequestClass:
+            request = RequestClass(**requestParams)
+            response = EndpointMethod(instance, *additionalParams, request)
+        else:
+            response = EndpointMethod(instance, *additionalParams)
+        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):
+        try:
+            return self.innerExecuteApiCall(ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams)
+        except defectdojo_api.exceptions.ApiException as e:
+            raise ApiException(e.status, e.reason, e.body, e.data, e.headers)
+
+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):
+        try:
+            return self.innerExecuteApiCall(ApiClass, EndpointMethod, RequestClass, requestParams, additionalParams)
+        except dependencytrack_api.exceptions.ApiException as e:
+            raise ApiException(e.status, e.reason, e.body, e.data, e.headers)
+
+
+
+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)
+
+
+
+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']
+        )
+        defectdojo_configuration.api_key['tokenAuth'] = config['DEFECTDOJO_TOKEN']
+        defectdojo_configuration.api_key_prefix['tokenAuth'] = 'Token'
+
+        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_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 = \
+                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 --------
+    dependencytrack_configuration = dependencytrack_api.Configuration(
+        host = f"{config['DTRACK_API_URL']}/api"
+    )
+    dependencytrack_configuration.debug = False
+    dependencytrack_configuration.api_key['ApiKeyAuth'] = config['DTRACK_TOKEN']
+
+    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_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:
+                client.executeApiCall(
+                    dependencytrack_api.ProjectPropertyApi,
+                    dependencytrack_api.ProjectPropertyApi.create_property1,
+                    dependencytrack_api.ProjectProperty,
+                    property,
+                    [ project_uuid ]
+                )
+
+        bom_response = \
+            client.executeApiCall(
+                dependencytrack_api.BomApi,
+                dependencytrack_api.BomApi.upload_bom,
+                None,
+                None,
+                [ None, False, projectName, projectVersion, None, None, None, None, True, sbom ]
+            )
+

src/sbom_dt_dd_api.py

@@ -0,0 +1,73 @@
+import os
+from loguru import logger
+from fastapi import FastAPI, UploadFile, File, Form, HTTPException
+from fastapi.responses import JSONResponse
+from converter import minimalSbomFormatConverter
+from sbom_dt_dd import generateSBOM, loadToDTrackAndDefectDojo, ApiException
+
+app = FastAPI(
+    title="SBOM DTrack DefectDojo Synchronization API",
+    version="0.0.1",
+    description=""
+)
+
+config = {}
+try:
+    config['DTRACK_API_URL'] = os.environ["DTRACK_API_URL"]
+    config['DTRACK_TOKEN'] = os.environ["DTRACK_TOKEN"]
+    config['DEFECTDOJO_URL'] = os.environ["DEFECTDOJO_URL"]
+    config['DEFECTDOJO_TOKEN'] = os.environ["DEFECTDOJO_TOKEN"]
+    config['VERBOSE'] = True
+except KeyError as e:
+    raise Exception(f"Env variable {e} is shall be set")
+
+app.state.config = config
+
+
+@app.get("/hello")
+async def say_hello(name: str):
+    """
+    Returns a friendly greeting.
+    ---
+    parameters:
+      - name: name
+        in: query
+        required: true
+        schema:
+          type: string
+    responses:
+      200:
+        description: Successful Response
+        content:
+          application/json:
+            schema:
+              type: object
+              properties:
+                message:
+                  type: string
+    """
+    return JSONResponse(content={"message": f"Hello, {name}!"})
+
+@app.post("/uploadMinimalSBOM/")
+async def uploadMinimalSBOM(
+    file: UploadFile = File(...),
+    reimport: bool = Form(...)
+):
+    """
+    Endpoint to upload a minimal SBOM definition
+    """
+    sbom = await file.read()
+
+    try:
+        logger.info("Start converting from minimal format into cyclonedx")
+        (sbom, projectName, projectVersion, projectClassifier, projectDescription) = minimalSbomFormatConverter(sbom)
+        logger.info("Converted")
+    
+        loadToDTrackAndDefectDojo(app.state.config, projectName, projectVersion, projectClassifier, projectDescription, 1, sbom, reimport)
+        logger.info("Done.")
+    except ApiException as e:
+        raise HTTPException(status_code=e.status, detail=f"{e.reason=}, {e.body=}, {e.data=}")
+
+    return JSONResponse(content={
+        "message": "Upload successful!"
+    })

src/sbom_dt_dd_cli.py

@@ -0,0 +1,133 @@
+import os
+import sys
+from loguru import logger
+import argparse
+import subprocess
+import json
+
+
+
+from converter import minimalSbomFormatConverter
+from sbom_dt_dd import generateSBOM, loadToDTrackAndDefectDojo
+
+
+
+# ---- main starts here with preparation of config -----------------------------------------------------------------------
+
+
+
+parser = argparse.ArgumentParser(description='sbom-dt-dd glue logic')
+parser.add_argument('--name', '-n',
+                    help='Project Name',
+                    required=False,
+                    default=''),
+parser.add_argument('--version', '-v',
+                    help='Project Version',
+                    required=False,
+                    default='')
+parser.add_argument('--description', '-d',
+                    help='Project Description',
+                    required=False,
+                    default='')
+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=False,
+                    default='')
+parser.add_argument('--uploadsbom', '-U',
+                    help='Upload a already existing SBOM instead of generating it. Give the SBOM file at -F instead of a target',
+                    required=False,
+                    action='store_true',
+                    default=False)
+parser.add_argument('--sbomfile', '-F',
+                    help='Filename of existing SBOM file to upload, use together with -U, do not use together with -T',
+                    required=False)
+parser.add_argument('--minimalsbomformat', '-K',
+                    help='SBOM file comes in dedicated minimal format and will be converted into cyclonedx before uploading',
+                    action='store_true',
+                    default=False)
+parser.add_argument('--overwritemetadata', '-O',
+                    help='Overwrite name, version, description and classifier with data from minimal SBOM',
+                    action='store_true',
+                    default=False)
+parser.add_argument('--target', '-T',
+                    help='Target to scan, either path name for sources or docker image tag',
+                    required=False)
+parser.add_argument('--reimport', '-R',
+                    help='Import the SBOM for an existing project/product once again',
+                    required=False,
+                    action='store_true',
+                    default=False)
+parser.add_argument('--verbose', '-V',
+                    help='A lot of debug output',
+                    required=False,
+                    action='store_true',
+                    default=False)
+args = parser.parse_args()
+projectName = args.name
+projectVersion = args.version
+projectDescription = args.description
+productType = args.type
+projectClassifier = args.classifier
+reImport = args.reimport
+
+uploadSbomFlag = args.uploadsbom
+if uploadSbomFlag:
+    sbomFileName = args.sbomfile
+    minimalSbomFormat = args.minimalsbomformat
+else:
+    target = args.target
+
+if minimalSbomFormat:
+    overwriteMetadata = args.overwritemetadata
+
+if not overwriteMetadata and not (projectName and projectVersion and projectClassifier and projectDescription):
+    raise MyLocalException("If overwriteMetadata is not selected, projectName, projectVersion, projectClassifier and projectDescription must be set.")
+
+
+CONFIG = {}
+try:
+    CONFIG['DTRACK_API_URL'] = os.environ["DTRACK_API_URL"]
+    CONFIG['DTRACK_TOKEN'] = os.environ["DTRACK_TOKEN"]
+    CONFIG['DEFECTDOJO_URL'] = os.environ["DEFECTDOJO_URL"]
+    CONFIG['DEFECTDOJO_TOKEN'] = os.environ["DEFECTDOJO_TOKEN"]
+except KeyError as e:
+    raise Exception(f"Env variable {e} is shall be set")
+
+CONFIG['VERBOSE'] = args.verbose
+
+# ---- main starts here --------------------------------------------------------------------------------------------------
+
+if uploadSbomFlag:
+    # ------- read uploaded SBOM -------------
+    logger.info(f"Reading SBOM from file {sbomFileName}")
+    with open(sbomFileName, 'r') as sbomFile:
+        sbom = sbomFile.read()
+    logger.info("SBOM file read.")
+    if minimalSbomFormat:
+        logger.info("Start converting from minimal format into cyclonedx")
+        (sbom, nameFromMinimalSbom, versionFromMinimalSbom, classifierFromMinimalSbom, descriptionFromMinimalSbom) = minimalSbomFormatConverter(sbom)
+        logger.info("Converted")
+        if overwriteMetadata:
+            projectName = nameFromMinimalSbom
+            projectVersion = versionFromMinimalSbom
+            projectClassifier = classifierFromMinimalSbom
+            projectDescription = descriptionFromMinimalSbom
+    logger.info("Done.")
+else:
+    # ------- generate SBOM ------------
+    logger.info(f"Generating SBOM for {target}")
+    sbomJson = generateSBOM(target, projectName, projectVersion)
+    sbom = json.dumps(sbomJson)
+    logger.info("Done.")
+
+
+
+loadToDTrackAndDefectDojo(CONFIG, projectName, projectVersion, projectClassifier, projectDescription, productType, sbom, reImport)
+
+

src/server.sh

@@ -0,0 +1,3 @@
+#!/bin/bash
+
+./.venv/bin/gunicorn sbom_dt_dd_api:app -k uvicorn.workers.UvicornWorker -w 4 -b 0.0.0.0:8000