feat(sbom):SP-4178 export acknowledgement as CycloneDX/SPDX annotations with timestamp and organization

Author: isasmendiagus

src/scanoss/cyclonedx.py

@@ -32,6 +32,7 @@
 from cyclonedx.validation.json import JsonValidator
 
 from . import __version__
+from .scanoss_settings import find_best_match
 from .scanossbase import ScanossBase
 from .spdxlite import SpdxLite
 
@@ -42,13 +43,14 @@ class CycloneDx(ScanossBase):
     Handle all interaction with CycloneDX formatting
     """
 
-    def __init__(self, debug: bool = False, output_file: str = None):
+    def __init__(self, debug: bool = False, output_file: str = None, scanoss_settings=None):
         """
         Initialise the CycloneDX class
         """
         super().__init__(debug)
         self.output_file = output_file
         self.debug = debug
+        self.scanoss_settings = scanoss_settings
         self._spdx = SpdxLite(debug=debug)
 
     def parse(self, data: dict):  # noqa: PLR0912, PLR0915
@@ -100,7 +102,7 @@ def parse(self, data: dict):  # noqa: PLR0912, PLR0915
                                     fdl.append({'id': name})
                                     dc.append(name)
                         fd['licenses'] = fdl
-                        fd['acknowledgement'] = deps.get('acknowledgement')
+                        fd['_file_path'] = f
                         cdx[purl] = fd
                 else:
                     purls = d.get('purl')
@@ -159,7 +161,7 @@ def parse(self, data: dict):  # noqa: PLR0912, PLR0915
                                 continue
                             fdl.append({'id': name})
                     fd['licenses'] = fdl
-                    fd['acknowledgement'] = d.get('acknowledgement')
+                    fd['_file_path'] = f
                     cdx[purl] = fd
         # self.print_stderr(f'VD: {vdx}')
         # self.print_stderr(f'CDX: {cdx}')
@@ -202,13 +204,12 @@ def produce_from_json(self, data: dict, output_file: str = None) -> tuple[bool,
             self.print_msg('Warning: Empty scan results - generating minimal CycloneDX SBOM with no components.')
         self._spdx.load_license_data()  # Load SPDX license name data for later reference
         #
-        # Using CDX version 1.4: https://cyclonedx.org/docs/1.4/json/
+        # Using CDX version 1.5: https://cyclonedx.org/docs/1.5/json/
         # Validate using: https://github.com/CycloneDX/cyclonedx-cli
-        # cyclonedx-cli validate --input-format json --input-version v1_4 --fail-on-errors --input-file cdx.json
         #
         data = {
             'bomFormat': 'CycloneDX',
-            'specVersion': '1.4',
+            'specVersion': '1.5',
             'serialNumber': f'urn:uuid:{uuid.uuid4()}',
             'version': 1,
             'metadata': {
@@ -255,11 +256,35 @@ def produce_from_json(self, data: dict, output_file: str = None) -> tuple[bool,
             cpe = comp.get('cpe', '')
             if cpe and cpe != '':
                 c_data['cpe'] = cpe
-            acknowledgement = comp.get('acknowledgement')
-            if acknowledgement:
-                c_data['properties'] = [{'name': 'scanoss:acknowledgement', 'value': acknowledgement}]
             data['components'].append(c_data)
         # End for loop
+        # Build annotations from BOM rules via ScanossSettings
+        annotations = []
+        if self.scanoss_settings:
+            all_entries = (self.scanoss_settings.get_bom_include()
+                           + self.scanoss_settings.get_bom_replace())
+            entries_with_ack = [e for e in all_entries if e.acknowledgement]
+            if entries_with_ack:
+                org = self.scanoss_settings.get_organization()
+                for purl in cdx:
+                    comp = cdx.get(purl)
+                    file_path = comp.get('_file_path', '')
+                    match = find_best_match(file_path, [purl], entries_with_ack)
+                    if match:
+                        ts = match.timestamp
+                        if not ts:
+                            self.print_stderr(
+                                f'Warning: No timestamp for annotation on {purl}, using current time'
+                            )
+                            ts = data['metadata']['timestamp']
+                        annotations.append({
+                            'subjects': [purl],
+                            'text': match.acknowledgement,
+                            'timestamp': ts,
+                            'annotator': {'organization': {'name': org}},
+                        })
+        if annotations:
+            data['annotations'] = annotations
         if vdx:
             for vuln_id in vdx:
                 vulns = vdx.get(vuln_id)

src/scanoss/scanner.py

@@ -612,10 +612,10 @@ def __finish_scan_threaded(self, file_map: Optional[Dict[Any, Any]] = None) -> b
         if self.output_format == 'plain':
             self.__log_result(json.dumps(results, indent=2, sort_keys=True))
         elif self.output_format == 'cyclonedx':
-            cdx = CycloneDx(self.debug, self.scan_output)
+            cdx = CycloneDx(self.debug, self.scan_output, scanoss_settings=self.scanoss_settings)
             success, _ = cdx.produce_from_json(results)
         elif self.output_format == 'spdxlite':
-            spdxlite = SpdxLite(self.debug, self.scan_output)
+            spdxlite = SpdxLite(self.debug, self.scan_output, scanoss_settings=self.scanoss_settings)
             success = spdxlite.produce_from_json(results)
         elif self.output_format == 'csv':
             csvo = CsvOutput(self.debug, self.scan_output)
@@ -1050,10 +1050,10 @@ def scan_wfp(self, wfp: str) -> bool:
         if self.output_format == 'plain':
             self.__log_result(raw_output)
         elif self.output_format == 'cyclonedx':
-            cdx = CycloneDx(self.debug, self.scan_output)
+            cdx = CycloneDx(self.debug, self.scan_output, scanoss_settings=self.scanoss_settings)
             cdx.produce_from_str(raw_output)
         elif self.output_format == 'spdxlite':
-            spdxlite = SpdxLite(self.debug, self.scan_output)
+            spdxlite = SpdxLite(self.debug, self.scan_output, scanoss_settings=self.scanoss_settings)
             success = spdxlite.produce_from_str(raw_output)
         elif self.output_format == 'csv':
             csvo = CsvOutput(self.debug, self.scan_output)

src/scanoss/scanoss_settings.py

@@ -46,6 +46,7 @@ class BomEntry:
     path: Optional[str] = None
     comment: Optional[str] = None
     acknowledgement: Optional[str] = None
+    timestamp: Optional[str] = None
 
     @classmethod
     def from_dict(cls, data: dict) -> 'BomEntry':
@@ -56,6 +57,7 @@ def from_dict(cls, data: dict) -> 'BomEntry':
             path=path,
             comment=data.get('comment'),
             acknowledgement=data.get('acknowledgement'),
+            timestamp=data.get('timestamp'),
         )
 
     def matches_path(self, result_path: str) -> bool:
@@ -112,6 +114,7 @@ def from_dict(cls, data: dict) -> 'ReplaceRule':
             path=path,
             comment=data.get('comment'),
             acknowledgement=data.get('acknowledgement'),
+            timestamp=data.get('timestamp'),
             replace_with=data.get('replace_with'),
             license=data.get('license'),
         )
@@ -321,6 +324,10 @@ def _get_bom(self):
                 return []
         return self.data.get('bom', {})
 
+    def get_organization(self) -> str:
+        """Get the organization name from self section. Returns 'unspecified' if not set."""
+        return self.data.get('self', {}).get('organization') or 'unspecified'
+
     def get_bom_include(self) -> List[BomEntry]:
         """
         Get the list of components to include in the scan

src/scanoss/spdxlite.py

@@ -34,6 +34,7 @@
 from packageurl import PackageURL
 
 from . import __version__
+from .scanoss_settings import find_best_match
 
 
 class SpdxLite:
@@ -42,12 +43,13 @@ class SpdxLite:
     Handle all interaction with SPDX Lite formatting
     """
 
-    def __init__(self, debug: bool = False, output_file: str = None):
+    def __init__(self, debug: bool = False, output_file: str = None, scanoss_settings=None):
         """
         Initialise the SpdxLite class
         """
         self.output_file = output_file
         self.debug = debug
+        self.scanoss_settings = scanoss_settings
         self._spdx_licenses = {}  # Used to lookup for valid SPDX license identifiers
         self._spdx_lic_names = {}  # Used to look for SPDX license identifiers by name
 
@@ -136,7 +138,9 @@ def _process_dependency_entry(self, file_path: str, entry: dict, summary: dict):
             if not self._is_valid_purl(file_path, dep, purl, summary):
                 continue
             # Modifying the summary dictionary directly as it's passed by reference
-            summary[purl] = self._create_dependency_summary(dep)
+            dep_summary = self._create_dependency_summary(dep)
+            dep_summary['_file_path'] = file_path
+            summary[purl] = dep_summary
 
     def _process_file_entry(self, file_path: str, entry: dict, summary: dict):
         """
@@ -156,7 +160,9 @@ def _process_file_entry(self, file_path: str, entry: dict, summary: dict):
         if not self._is_valid_purl(file_path, entry, purl, summary):
             return
 
-        summary[purl] = self._create_file_summary(entry)
+        file_summary = self._create_file_summary(entry)
+        file_summary['_file_path'] = file_path
+        summary[purl] = file_summary
 
     def _is_valid_purl(self, file_path: str, entry: dict, purl: str, summary: dict) -> bool:
         """
@@ -199,7 +205,6 @@ def _create_dependency_summary(self, dep: dict) -> dict:
         for field in ['component', 'version', 'url']:
             summary[field] = dep.get(field, '')
         summary['licenses'] = self._process_licenses(dep.get('licenses'))
-        summary['acknowledgement'] = dep.get('acknowledgement')
         return summary
 
     def _create_file_summary(self, entry: dict) -> dict:
@@ -220,7 +225,6 @@ def _create_file_summary(self, entry: dict) -> dict:
         for field in fields:
             summary[field] = entry.get(field)
         summary['licenses'] = self._process_licenses(entry.get('licenses'))
-        summary['acknowledgement'] = entry.get('acknowledgement')
         return summary
 
     def _process_licenses(self, licenses: list) -> list:
@@ -293,6 +297,7 @@ def produce_from_json(self, data: json, output_file: str = None) -> bool:
         self.load_license_data()
         spdx_document = self._create_base_document(raw_data)
         self._process_packages(raw_data, spdx_document)
+        self._build_annotations(raw_data, spdx_document)
         return self._write_output(spdx_document, output_file)
 
     def _create_base_document(self, raw_data: dict) -> dict:
@@ -392,6 +397,36 @@ def _process_packages(self, raw_data: dict, spdx_document: dict):
 
         self._process_license_refs(lic_refs, spdx_document)
 
+    def _build_annotations(self, raw_data: dict, spdx_document: dict):
+        """Build SPDX annotations from BOM rules via ScanossSettings."""
+        if not self.scanoss_settings:
+            return
+        all_entries = (self.scanoss_settings.get_bom_include()
+                       + self.scanoss_settings.get_bom_replace())
+        entries_with_ack = [e for e in all_entries if e.acknowledgement]
+        if not entries_with_ack:
+            return
+        annotations = []
+        org = self.scanoss_settings.get_organization()
+        for purl, comp in raw_data.items():
+            file_path = comp.get('_file_path', '')
+            match = find_best_match(file_path, [purl], entries_with_ack)
+            if match:
+                ts = match.timestamp
+                if not ts:
+                    self.print_stderr(
+                        f'Warning: No timestamp for annotation on {purl}, using current time'
+                    )
+                    ts = spdx_document['creationInfo']['created']
+                annotations.append({
+                    'annotationDate': ts,
+                    'annotationType': 'REVIEW',
+                    'annotator': f'Organization: {org}',
+                    'comment': match.acknowledgement,
+                })
+        if annotations:
+            spdx_document['annotations'] = annotations
+
     def _create_package_info(self, purl: str, comp: dict, lic_refs: set) -> dict:
         """
             Create package information for SPDX document.
@@ -453,9 +488,6 @@ def _create_package_info(self, purl: str, comp: dict, lic_refs: set) -> dict:
                 }
             ],
         }
-        acknowledgement = comp.get('acknowledgement')
-        if acknowledgement:
-            package_info['comment'] = acknowledgement
         return package_info
 
     def _process_package_licenses(self, licenses: list, lic_refs: set) -> str: