feat(vex): Refactered generate.py and fixed tests

Author: JigyasuRajput

cve_bin_tool/vex_manager/generate.py

@@ -1,15 +1,13 @@
-# Copyright (C) 2024 Intel Corporation
+# Copyright (C) 2025 Intel Corporation
 # SPDX-License-Identifier: GPL-3.0-or-later
 
 from logging import Logger
 from pathlib import Path
 from typing import Dict, List, Optional
 
-from lib4sbom.data.vulnerability import Vulnerability
-from lib4vex.generator import VEXGenerator
-
 from cve_bin_tool.log import LOGGER
 from cve_bin_tool.util import CVEData, ProductInfo, Remarks
+from cve_bin_tool.vex_manager.handler import VexHandler
 
 
 class VEXGenerate:
@@ -103,40 +101,50 @@ def __init__(
         self.all_cve_data = all_cve_data
         self.sbom_serial_number = sbom_serial_number
 
+        # Initialize the VexHandler for generation operations
+        self.vex_handler = VexHandler(self.logger)
+
     def generate_vex(self) -> None:
         """
         Generates a VEX (Vulnerability Exploitability eXchange) document based on the specified VEX type
         and stores it in the given filename.
 
-        This method sets up a VEX generator instance with the product name, release version, and other
-        metadata. It automatically assigns a filename if none is provided, logs the update status if the
-        file already exists, and generates the VEX document with product vulnerability data.
+        This method delegates to the VexHandler for the actual generation, after preparing the data
+        structure with product, vulnerabilities and metadata. It automatically assigns a filename if
+        none is provided and logs update status if the file already exists.
 
         Returns:
             None
         """
-        author = "Unknown Author"
-        if self.vendor:
-            author = self.vendor
-        vexgen = VEXGenerator(vex_type=self.vextype, author=author)
-        kwargs = {"name": self.product, "release": self.release}
-        if self.sbom:
-            kwargs["sbom"] = self.sbom
-        vexgen.set_product(**kwargs)
         if not self.filename:
-            self.logger.info(
+            self.logger.debug(
                 "No filename defined, generating a new filename with default naming convention."
             )
             self.filename = self.__generate_vex_filename()
+
         if Path(self.filename).is_file():
-            self.logger.info(f"Updating the VEX file: {self.filename}")
+            self.logger.debug(f"Updating the VEX file: {self.filename}")
+
+        # Prepare data structure for VexHandler
+        vex_data = {
+            "product": {
+                "name": self.product,
+                "release": self.release,
+                "vendor": self.vendor,
+            },
+            "project_name": self.product,
+            "vulnerabilities": self.__get_vulnerabilities(),
+            "metadata": self.__get_metadata(),
+        }
+
+        # Add SBOM if available
+        if self.sbom:
+            vex_data["sbom"] = self.sbom
 
-        vexgen.generate(
-            project_name=self.product,
-            vex_data=self.__get_vulnerabilities(),
-            metadata=self.__get_metadata(),
-            filename=self.filename,
-        )
+        # Generate VEX document using the handler
+        success = self.vex_handler.generate(vex_data, self.filename, self.vextype)
+        if not success:
+            self.logger.error(f"Failed to generate VEX file: {self.filename}")
 
     def __generate_vex_filename(self) -> str:
         """
@@ -183,56 +191,62 @@ def __get_metadata(self) -> Dict:
 
         return metadata
 
-    def __get_vulnerabilities(self) -> List[Vulnerability]:
+    def __get_vulnerabilities(self) -> List[Dict]:
         """
-        Retrieves and constructs a list of vulnerability objects based on the current CVE data.
+        Prepares a list of vulnerability dictionaries for the VEX document based on the current CVE data.
 
         This method iterates through all CVE data associated with the product and vendor,
-        creating and configuring `Vulnerability` objects for each entry. It sets attributes
-        like name, release, ID, description, status, and additional metadata such as package
-        URLs (purl) and bill of materials (BOM) links. If a vulnerability includes comments
-        or justification, these are added to the vulnerability details.
+        creating vulnerability dictionaries for each entry with attributes like ID, description,
+        status, and additional metadata such as package URLs (purl) and bill of materials (BOM) links.
 
         Returns:
-            List[Vulnerability]: A list of `Vulnerability` objects representing the identified
-            vulnerabilities, enriched with metadata and details.
+            List[Dict]: A list of vulnerability dictionaries ready for VexHandler to process.
         """
         vulnerabilities = []
         for product_info, cve_data in self.all_cve_data.items():
             vendor, product, version, _, purl = product_info
             for cve in cve_data["cves"]:
                 if isinstance(cve, str):
                     continue
-                vulnerability = Vulnerability(validation=self.vextype)
-                vulnerability.initialise()
-                vulnerability.set_name(product)
-                vulnerability.set_release(version)
-                vulnerability.set_id(cve.cve_number)
-                vulnerability.set_description(cve.description)
-                vulnerability.set_comment(cve.comments)
-                vulnerability.set_status(self.analysis_state[self.vextype][cve.remarks])
+
+                # Create a vulnerability dictionary in the format expected by VexHandler
+                vulnerability = {
+                    "name": product,
+                    "release": version,
+                    "id": cve.cve_number,
+                    "description": cve.description,
+                    "comment": cve.comments,
+                    "status": self.analysis_state[self.vextype][cve.remarks],
+                }
+
                 if cve.justification:
-                    vulnerability.set_justification(cve.justification)
-                if cve.response:
-                    vulnerability.set_value("remediation", cve.response[0])
+                    vulnerability["justification"] = cve.justification
+
+                if cve.response and len(cve.response) > 0:
+                    vulnerability["remediation"] = cve.response[0]
+
                 detail = (
                     f"{cve.remarks.name}: {cve.comments}"
                     if cve.comments
                     else cve.remarks.name
                 )
+
                 if purl is None:
                     purl = f"pkg:generic/{vendor}/{product}@{version}"
+
                 bom_version = 1
                 if self.sbom_serial_number != "":
                     ref = f"urn:cdx:{self.sbom_serial_number}/{bom_version}#{purl}"
                 else:
                     ref = f"urn:cbt:{bom_version}/{vendor}#{product}:{version}"
 
-                vulnerability.set_value("purl", str(purl))
-                vulnerability.set_value("bom_link", ref)
-                vulnerability.set_value("action", detail)
-                vulnerability.set_value("source", cve.data_source)
-                vulnerability.set_value("updated", cve.last_modified)
-                vulnerabilities.append(vulnerability.get_vulnerability())
+                vulnerability["purl"] = str(purl)
+                vulnerability["bom_link"] = ref
+                vulnerability["action"] = detail
+                vulnerability["source"] = cve.data_source
+                vulnerability["updated"] = cve.last_modified
+
+                vulnerabilities.append(vulnerability)
+
         self.logger.debug(f"Vulnerabilities: {vulnerabilities}")
         return vulnerabilities

cve_bin_tool/vex_manager/handler.py

@@ -10,7 +10,27 @@
 
 from cve_bin_tool.log import LOGGER
 from cve_bin_tool.util import ProductInfo, Remarks, decode_bom_ref, decode_purl
-from lib4vex import Validator as VEXValidator
+
+# Import VEX validator with proper error handling for open source compatibility
+try:
+    # Try the most common import path first
+    from lib4vex import Validator as VEXValidator
+except ImportError:
+    try:
+        # Try alternative import paths
+        from lib4vex.validator import VEXValidator
+    except ImportError:
+        try:
+            from lib4vex.validate import Validator as VEXValidator
+        except ImportError:
+            # If no validator is available, create a clear error message
+            VEXValidator = None
+            VALIDATOR_IMPORT_ERROR = (
+                "VEX validation functionality is not available. "
+                "This may be due to an incompatible version of lib4vex. "
+                "Please check that lib4vex is properly installed and up to date. "
+                "Validation methods will be disabled but parsing and generation will still work."
+            )
 
 TriageData = Dict[str, Union[Dict[str, Any], Set[str]]]
 
@@ -63,6 +83,10 @@ def __init__(self, logger=None):
         """
         self.logger = logger or LOGGER.getChild(self.__class__.__name__)
 
+        # Warn user if validator is not available
+        if VEXValidator is None:
+            self.logger.warning(VALIDATOR_IMPORT_ERROR)
+
     def parse(
         self, filename: str, vextype: str = "auto"
     ) -> DefaultDict[ProductInfo, TriageData]:
@@ -88,7 +112,7 @@ def parse(
             if vextype == "auto":
                 vextype = vexparser.get_type()
 
-            self.logger.info(f"Parsed VEX file: {filename} of type: {vextype}")
+            self.logger.debug(f"Parsed VEX file: {filename} of type: {vextype}")
 
             return self._process_parsed_data(vexparser, vextype)
 
@@ -111,12 +135,20 @@ def validate(self, filename: str, vextype: str = "auto") -> bool:
             self.logger.error(f"VEX file not found: {filename}")
             return False
 
+        # Check if validator is available
+        if VEXValidator is None:
+            self.logger.error(
+                "VEX validation is not available. "
+                "Please ensure lib4vex is properly installed with validation support."
+            )
+            return False
+
         try:
             validator = VEXValidator(vex_type=vextype)
             is_valid = validator.validate(filename)
 
             if is_valid:
-                self.logger.info(f"VEX file {filename} is valid")
+                self.logger.debug(f"VEX file {filename} is valid")
             else:
                 self.logger.error(f"VEX file {filename} is invalid")
 
@@ -141,7 +173,7 @@ def generate(self, data: Dict, output_file: str, vextype: str) -> bool:
         try:
             generator = VEXGenerator(vex_type=vextype)
             generator.generate(data, output_file)
-            self.logger.info(f"Generated {vextype} VEX file: {output_file}")
+            self.logger.debug(f"Generated {vextype} VEX file: {output_file}")
             return True
 
         except Exception as e: