Merge pull request #133 from scanoss/feature/mdaloia/add-vulns-to-cdx-hfh-output

feat: add vulnerabilities into cdx output when using folder hashing command

Author: matiasdaloia

CHANGELOG.md

@@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 ### Added
 - Upcoming changes...
 
+## [1.28.0] - 2025-07-10
+### Added
+- Add vulnerabilities response to `folder-scan` CycloneDX output
+
 ## [1.27.1] - 2025-07-09
 ### Fixed
 - Fixed when running `folder-scan` with `--format cyclonedx` the output was not writing to file

src/scanoss/__init__.py

@@ -22,4 +22,4 @@
   THE SOFTWARE.
 """
 
-__version__ = '1.27.1'
+__version__ = '1.28.0'

src/scanoss/cyclonedx.py

@@ -287,6 +287,87 @@ def produce_from_str(self, json_str: str, output_file: str = None) -> bool:
             return False
         return self.produce_from_json(data, output_file)
 
+    def _normalize_vulnerability_id(self, vuln: dict) -> tuple[str, str]:
+        """
+        Normalize vulnerability ID and CVE from different possible field names.
+        Returns tuple of (vuln_id, vuln_cve).
+        """
+        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.
+        """
+        vuln_source = vuln.get('source', '').lower()
+        return {
+            'id': vuln_id,
+            '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}'
+            },
+            'ratings': [{'severity': self._sev_lookup(vuln.get('severity', 'unknown').lower())}],
+            '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
+            )
+            
+            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)
+                )
+                    
+        return cdx_dict
+
     @staticmethod
     def _sev_lookup(value: str):
         """

src/scanoss/scanners/scanner_hfh.py

@@ -193,8 +193,13 @@ def _format_cyclonedx_output(self) -> str:  # noqa: PLR0911
                     }
                 ]
             }
+            
+            get_vulnerabilities_json_request = {
+                'purls': [{'purl': purl, 'requirement': best_match_version['version']}],
+            }
 
             decorated_scan_results = self.scanner.client.get_dependencies(get_dependencies_json_request)
+            vulnerabilities = self.scanner.client.get_vulnerabilities_json(get_vulnerabilities_json_request)
 
             cdx = CycloneDx(self.base.debug)
             scan_results = {}
@@ -205,6 +210,10 @@ def _format_cyclonedx_output(self) -> str:  # noqa: PLR0911
                 error_msg = 'ERROR: Failed to produce CycloneDX output'
                 self.base.print_stderr(error_msg)
                 return None
+            
+            if vulnerabilities:
+                cdx_output = cdx.append_vulnerabilities(cdx_output, vulnerabilities, purl)
+                
             return json.dumps(cdx_output, indent=2)
         except Exception as e:
             self.base.print_stderr(f'ERROR: Failed to get license information: {e}')

src/scanoss/scanossgrpc.py

@@ -326,7 +326,7 @@ def get_vulnerabilities_json(self, purls: dict) -> dict:
             request = ParseDict(purls, PurlRequest())  # Parse the JSON/Dict into the purl request object
             metadata = self.metadata[:]
             metadata.append(('x-request-id', request_id))  # Set a Request ID
-            self.print_debug(f'Sending crypto data for decoration (rqId: {request_id})...')
+            self.print_debug(f'Sending vulnerability data for decoration (rqId: {request_id})...')
             resp = self.vuln_stub.GetVulnerabilities(request, metadata=metadata, timeout=self.timeout)
         except Exception as e:
             self.print_stderr(