[SP-2879] feat: add export dt sub-command, add cyclonedx input file validation

Author: matiasdaloia

CHANGELOG.md

@@ -9,6 +9,11 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Upcoming changes...
 
+## [1.30.0] - 2025-07-22
+### Added
+- Add `export dt` subcommand to export SBOM files to Dependency Track
+- Add CycloneDX file validation
+
 ## [1.29.0] - 2025-07-15
 ### Changed
 - Updated minimum Python version to 3.9
@@ -609,4 +614,6 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 [1.28.1]: https://github.com/scanoss/scanoss.py/compare/v1.28.0...v1.28.1
 [1.28.2]: https://github.com/scanoss/scanoss.py/compare/v1.28.1...v1.28.2
 [1.29.0]: https://github.com/scanoss/scanoss.py/compare/v1.28.2...v1.29.0
+[1.30.0]: https://github.com/scanoss/scanoss.py/compare/v1.29.0...v1.30.0
+
 

requirements.txt

@@ -12,4 +12,6 @@ importlib_resources
 packageurl-python
 pathspec
 jsonschema
-crc
+crc
+
+cyclonedx-python-lib[validation]

setup.cfg

@@ -39,6 +39,7 @@ install_requires =
     pathspec
     jsonschema
     crc
+    cyclonedx-python-lib[validation]
 
 
 [options.extras_require]

src/scanoss/__init__.py

@@ -22,4 +22,4 @@
   THE SOFTWARE.
 """
 
-__version__ = '1.29.0'
+__version__ = '1.30.0'

src/scanoss/cli.py

@@ -25,13 +25,18 @@
 import argparse
 import os
 import sys
+import traceback
 from dataclasses import asdict
 from pathlib import Path
 from typing import List
 
 import pypac
 
 from scanoss.cryptography import Cryptography, create_cryptography_config_from_args
+from scanoss.export.dependency_track import (
+    DependencyTrackExporter,
+    create_dependency_track_exporter_config_from_args,
+)
 from scanoss.inspection.component_summary import ComponentSummary
 from scanoss.inspection.license_summary import LicenseSummary
 from scanoss.scanners.container_scanner import (
@@ -553,13 +558,17 @@ def setup_args() -> None:  # noqa: PLR0912, PLR0915
     ####### INSPECT: License Summary ######
     # Inspect Sub-command: inspect license summary
     p_license_summary = p_inspect_sub.add_parser(
-        'license-summary', aliases=['lic-summary', 'licsum'], description='Get license summary',
-        help='Get detected license summary from scan results'
+        'license-summary',
+        aliases=['lic-summary', 'licsum'],
+        description='Get license summary',
+        help='Get detected license summary from scan results',
     )
 
     p_component_summary = p_inspect_sub.add_parser(
-        'component-summary', aliases=['comp-summary', 'compsum'], description='Get component summary',
-        help='Get detected component summary from scan results'
+        'component-summary',
+        aliases=['comp-summary', 'compsum'],
+        description='Get component summary',
+        help='Get detected component summary from scan results',
     )
 
     ####### INSPECT: Undeclared components ######
@@ -605,6 +614,36 @@ def setup_args() -> None:  # noqa: PLR0912, PLR0915
 
     ########################################### END INSPECT SUBCOMMAND ###########################################
 
+    # Sub-command: export
+    p_export = subparsers.add_parser(
+        'export',
+        aliases=['exp'],
+        description=f'Export SBOM files to external platforms: {__version__}',
+        help='Export SBOM files to external platforms',
+    )
+
+    export_sub = p_export.add_subparsers(
+        title='Export Commands',
+        dest='subparsercmd',
+        description='export sub-commands',
+        help='export sub-commands',
+    )
+
+    # Export Sub-command: export dt (Dependency Track)
+    e_dt = export_sub.add_parser(
+        'dt',
+        aliases=['dependency-track'],
+        description='Export SBOM to Dependency Track',
+        help='Upload SBOM files to Dependency Track',
+    )
+    e_dt.add_argument('-i', '--input', type=str, required=True, help='Input SBOM file (CycloneDX JSON format)')
+    e_dt.add_argument('--dt-url', type=str, required=True, help='Dependency Track base URL')
+    e_dt.add_argument('--dt-apikey', type=str, required=True, help='Dependency Track API key')
+    e_dt.add_argument('--dt-projectid', type=str, help='Dependency Track project UUID')
+    e_dt.add_argument('--dt-projectname', type=str, help='Dependency Track project name')
+    e_dt.add_argument('--dt-projectversion', type=str, help='Dependency Track project version')
+    e_dt.set_defaults(func=export_dt)
+
     # Sub-command: folder-scan
     p_folder_scan = subparsers.add_parser(
         'folder-scan',
@@ -858,6 +897,7 @@ def setup_args() -> None:  # noqa: PLR0912, PLR0915
         p_crypto_algorithms,
         p_crypto_hints,
         p_crypto_versions_in_range,
+        e_dt,
     ]:
         p.add_argument('--debug', '-d', action='store_true', help='Enable debug messages')
         p.add_argument('--trace', '-t', action='store_true', help='Enable trace messages, including API posts')
@@ -871,7 +911,8 @@ def setup_args() -> None:  # noqa: PLR0912, PLR0915
         parser.print_help()  # No sub command subcommand, print general help
         sys.exit(1)
     elif (
-        args.subparser in ('utils', 'ut', 'component', 'comp', 'inspect', 'insp', 'ins', 'crypto', 'cr')
+        args.subparser
+        in ('utils', 'ut', 'component', 'comp', 'inspect', 'insp', 'ins', 'crypto', 'cr', 'export', 'exp')
     ) and not args.subparsercmd:
         parser.parse_args([args.subparser, '--help'])  # Force utils helps to be displayed
         sys.exit(1)
@@ -1304,6 +1345,7 @@ def convert(parser, args):
     if not success:
         sys.exit(1)
 
+
 ################################ INSPECT handlers ################################
 def inspect_copyleft(parser, args):
     """
@@ -1381,16 +1423,17 @@ def inspect_undeclared(parser, args):
     status, _ = i_undeclared.run()
     sys.exit(status)
 
+
 def inspect_license_summary(parser, args):
     """
-       Run the "inspect" sub-command
-       Parameters
-       ----------
-           parser: ArgumentParser
-               command line parser object
-           args: Namespace
-               Parsed arguments
-       """
+    Run the "inspect" sub-command
+    Parameters
+    ----------
+        parser: ArgumentParser
+            command line parser object
+        args: Namespace
+            Parsed arguments
+    """
     if args.input is None:
         print_stderr('Please specify an input file to inspect')
         parser.parse_args([args.subparser, args.subparsercmd, '-h'])
@@ -1412,16 +1455,17 @@ def inspect_license_summary(parser, args):
     )
     i_license_summary.run()
 
+
 def inspect_component_summary(parser, args):
     """
-       Run the "inspect" sub-command
-       Parameters
-       ----------
-           parser: ArgumentParser
-               command line parser object
-           args: Namespace
-               Parsed arguments
-       """
+    Run the "inspect" sub-command
+    Parameters
+    ----------
+        parser: ArgumentParser
+            command line parser object
+        args: Namespace
+            Parsed arguments
+    """
     if args.input is None:
         print_stderr('Please specify an input file to inspect')
         parser.parse_args([args.subparser, args.subparsercmd, '-h'])
@@ -1440,8 +1484,42 @@ def inspect_component_summary(parser, args):
     )
     i_component_summary.run()
 
+
 ################################ End inspect handlers ################################
 
+
+def export_dt(parser, args):
+    """
+    Run the "export dt" sub-command
+    Parameters
+    ----------
+        parser: ArgumentParser
+            command line parser object
+        args: Namespace
+            Parsed arguments
+    """
+
+    try:
+        config = create_dependency_track_exporter_config_from_args(args)
+        dt_exporter = DependencyTrackExporter(
+            config=config,
+            debug=args.debug,
+            trace=args.trace,
+            quiet=args.quiet,
+        )
+
+        success = dt_exporter.upload_sbom(args.input)
+
+        if not success:
+            sys.exit(1)
+
+    except Exception as e:
+        print_stderr(f'ERROR: {e}')
+        if args.debug:
+            traceback.print_exc()
+        sys.exit(1)
+
+
 def utils_certloc(*_):
     """
     Run the "utils certloc" sub-command

src/scanoss/cyclonedx.py

@@ -28,6 +28,9 @@
 import sys
 import uuid
 
+from cyclonedx.schema import SchemaVersion
+from cyclonedx.validation.json import JsonValidator
+
 from . import __version__
 from .scanossbase import ScanossBase
 from .spdxlite import SpdxLite
@@ -296,13 +299,13 @@ def _normalize_vulnerability_id(self, vuln: dict) -> tuple[str, str]:
         """
         vuln_id = vuln.get('ID', '') or vuln.get('id', '')
         vuln_cve = vuln.get('CVE', '') or vuln.get('cve', '')
-        
+
         # Skip CPE entries, use CVE if available
         if vuln_id.upper().startswith('CPE:') and vuln_cve:
             vuln_id = vuln_cve
-        
+
         return vuln_id, vuln_cve
-    
+
     def _create_vulnerability_entry(self, vuln_id: str, vuln: dict, vuln_cve: str, purl: str) -> dict:
         """
         Create a new vulnerability entry for CycloneDX format.
@@ -313,61 +316,56 @@ def _create_vulnerability_entry(self, vuln_id: str, vuln: dict, vuln_cve: str, p
             'source': {
                 'name': 'NVD' if vuln_source == 'nvd' else 'GitHub Advisories',
                 'url': f'https://nvd.nist.gov/vuln/detail/{vuln_cve}'
-                      if vuln_source == 'nvd'
-                      else f'https://github.com/advisories/{vuln_id}'
+                if vuln_source == 'nvd'
+                else f'https://github.com/advisories/{vuln_id}',
             },
             'ratings': [{'severity': self._sev_lookup(vuln.get('severity', 'unknown').lower())}],
-            'affects': [{'ref': purl}]
+            'affects': [{'ref': purl}],
         }
-    
+
     def append_vulnerabilities(self, cdx_dict: dict, vulnerabilities_data: dict, purl: str) -> dict:
         """
         Append vulnerabilities to an existing CycloneDX dictionary
-        
+
         Args:
             cdx_dict (dict): The existing CycloneDX dictionary
             vulnerabilities_data (dict): The vulnerabilities data from get_vulnerabilities_json
             purl (str): The PURL of the component these vulnerabilities affect
-            
+
         Returns:
             dict: The updated CycloneDX dictionary with vulnerabilities appended
         """
         if not cdx_dict or not vulnerabilities_data:
             return cdx_dict
-            
+
         if 'vulnerabilities' not in cdx_dict:
             cdx_dict['vulnerabilities'] = []
-            
+
         # Extract vulnerabilities from the response
         vulns_list = vulnerabilities_data.get('purls', [])
         if not vulns_list:
             return cdx_dict
-            
+
         vuln_items = vulns_list[0].get('vulnerabilities', [])
-        
+
         for vuln in vuln_items:
             vuln_id, vuln_cve = self._normalize_vulnerability_id(vuln)
-            
+
             # Skip empty IDs or CPE-only entries
             if not vuln_id or vuln_id.upper().startswith('CPE:'):
                 continue
-            
+
             # Check if vulnerability already exists
-            existing_vuln = next(
-                (v for v in cdx_dict['vulnerabilities'] if v.get('id') == vuln_id),
-                None
-            )
-            
+            existing_vuln = next((v for v in cdx_dict['vulnerabilities'] if v.get('id') == vuln_id), None)
+
             if existing_vuln:
                 # Add this PURL to the affects list if not already present
                 if not any(ref.get('ref') == purl for ref in existing_vuln.get('affects', [])):
                     existing_vuln['affects'].append({'ref': purl})
             else:
                 # Create new vulnerability entry
-                cdx_dict['vulnerabilities'].append(
-                    self._create_vulnerability_entry(vuln_id, vuln, vuln_cve, purl)
-                )
-                    
+                cdx_dict['vulnerabilities'].append(self._create_vulnerability_entry(vuln_id, vuln, vuln_cve, purl))
+
         return cdx_dict
 
     @staticmethod
@@ -388,6 +386,25 @@ def _sev_lookup(value: str):
             'unknown': 'unknown',
         }.get(value, 'unknown')
 
+    def is_cyclonedx_json(self, json_string: str) -> bool:
+        """
+        Validate if the given JSON string is a valid CycloneDX JSON string
+        Args:
+            json_string (str): JSON string to validate
+        Returns:
+            bool: True if the JSON string is valid, False otherwise
+        """
+        try:
+            cdx_json_validator = JsonValidator(SchemaVersion.V1_6)
+            json_validation_errors = cdx_json_validator.validate_str(json_string)
+            if json_validation_errors:
+                self.print_stderr(f'ERROR: Problem parsing input JSON: {json_validation_errors}')
+                return False
+            return True
+        except Exception as e:
+            self.print_stderr(f'ERROR: Problem parsing input JSON: {e}')
+            return False
+
 
 #
 # End of CycloneDX Class