Skip to content
Closed
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
48 changes: 35 additions & 13 deletions cve_bin_tool/output_engine/util.py
Original file line number Diff line number Diff line change
Expand Up @@ -273,8 +273,8 @@ def intermediate_output(

def add_extension_if_not(filename: str, output_type: str) -> str:
"""
summary: Checks if the filename ends with the extension and if not
adds one. And if the filename ends with a different extension it replaces the extension.
Handles both replacement of invalid extensions (for known types)
and appending for completely unknown extensions.

Args:
filename (str): filename from OutputEngine
Expand All @@ -283,18 +283,40 @@ def add_extension_if_not(filename: str, output_type: str) -> str:
Returns:
str: Filename with extension according to output_type
"""
import re
# Map all output types to their valid extensions
extensions = {
"json": ["json"],
"cyclonedx": ["json", "xml"],
"csv": ["csv"],
"html": ["html"],
"pdf": ["pdf"],
"txt": ["txt"],
}

# Create set of ALL valid extensions for recognition
all_valid_extensions = {ext for exts in extensions.values() for ext in exts}

# Get valid extensions for current output type
valid_ext = extensions.get(output_type, [])

extensions = ["json", "csv", "html", "pdf", "txt"]
for extension in extensions:
if not filename.endswith(f".{extension}"):
continue
if extension == output_type:
# Split filename
if "." in filename:
name, ext = filename.rsplit(".", 1)
# Check if extension is either:
# 1. Valid for current type -> keep
# 2. Valid for another type -> replace
# 3. Invalid everywhere -> append
if ext in valid_ext:
return filename
filename = re.sub(f".{extension}$", f".{output_type}", filename)
return filename
filename = f"{filename}.{output_type}"
return filename
elif ext in all_valid_extensions:
# Replace with first valid extension for current type
return f"{name}.{valid_ext[0]}"
else:
# Append first valid extension for current type
return f"{filename}.{valid_ext[0]}"
else:
# No extension - append first valid one
return f"{filename}.{valid_ext[0]}" if valid_ext else filename


def group_cve_by_remark(
Expand All @@ -308,7 +330,7 @@ def group_cve_by_remark(
{
"cve_number": "CVE-XXX-XXX",
"severity": "High",
"decription: "Lorem Ipsm",
"description: "Lorem Ipsm",
},
{...}
],
Expand Down
5 changes: 5 additions & 0 deletions cve_bin_tool/sbom_manager/generate.py
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
from lib4sbom.sbom import SBOM

from cve_bin_tool.log import LOGGER
from cve_bin_tool.output_engine.util import add_extension_if_not
from cve_bin_tool.version import VERSION


Expand Down Expand Up @@ -46,6 +47,10 @@ def __init__(

def generate_sbom(self) -> None:
"""Create SBOM package and generate SBOM file."""
# Force .json extension for CycloneDX only if not already specified
if self.sbom_type == "cyclonedx":
self.filename = add_extension_if_not(self.filename, "json")

# Create SBOM
sbom_relationships = []
my_package = SBOMPackage()
Expand Down
67 changes: 43 additions & 24 deletions cve_bin_tool/sbom_manager/parse.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,6 +3,7 @@

from __future__ import annotations

import json
import re
import sys
from collections import defaultdict
Expand All @@ -16,13 +17,7 @@
from cve_bin_tool.cvedb import CVEDB
from cve_bin_tool.input_engine import TriageData
from cve_bin_tool.log import LOGGER
from cve_bin_tool.util import (
ProductInfo,
Remarks,
decode_cpe22,
decode_cpe23,
validate_serialNumber,
)
from cve_bin_tool.util import ProductInfo, Remarks, decode_cpe22, decode_cpe23
from cve_bin_tool.validator import validate_cyclonedx, validate_spdx, validate_swid


Expand Down Expand Up @@ -77,12 +72,22 @@ def parse_sbom(self) -> dict[ProductInfo, TriageData]:
modules = []
try:
if Path(self.filename).exists():
# Validate CycloneDX JSON or XML extension
if self.type == "cyclonedx" and not (
self.filename.lower().endswith(".json")
or self.filename.lower().endswith(".xml")
):
self.logger.error(
"CycloneDX SBOMs require .json or .xml extension."
)
return {}

if self.type == "swid":
modules = self.parse_swid(self.filename)
else:
modules = self.parse_cyclonedx_spdx()
except (KeyError, FileNotFoundError, ET.ParseError) as e:
LOGGER.debug(e, exc_info=True)
self.logger.debug(e, exc_info=True)

LOGGER.debug(
f"The number of modules identified in SBOM - {len(modules)}\n{modules}"
Expand Down Expand Up @@ -147,7 +152,7 @@ def common_prefix_split(self, product, version) -> list[ProductInfo]:
if not found_common_prefix:
# if vendor not found after removing common prefix try splitting it
LOGGER.debug(
f"No Vendor found for {product}, trying splitted product. "
f"No Vendor found for {product}, trying split product. "
"Some results may be inaccurate due to vendor identification limitations."
)
splitted_product = product.split("-")
Expand Down Expand Up @@ -217,31 +222,45 @@ def parse_cyclonedx_spdx(self) -> [(str, str, str)]:

Returns:
- List[(str, str, str)]: A list of tuples, each containing vendor, product, and version information for a module.

"""
# Validate CycloneDX JSON or XML extension
if self.type == "cyclonedx" and not (
self.filename.lower().endswith(".json")
or self.filename.lower().endswith(".xml")
):
self.logger.error(
f"CycloneDX SBOMs require .json or .xml extension. Invalid file: {self.filename}"
)
return []

# Validate JSON content for CycloneDX JSON files
if self.type == "cyclonedx" and self.filename.lower().endswith(".json"):
try:
with open(self.filename, encoding="utf-8") as f:
json.load(f) # Basic JSON validation
except json.JSONDecodeError as e:
self.logger.error(f"Invalid JSON in CycloneDX SBOM: {str(e)}")
return []

# Set up SBOM parser
sbom_parser = SBOMParser(sbom_type=self.type)
# Load SBOM
sbom_parser.parse_file(self.filename)
doc = sbom_parser.get_document()
uuid = doc.get("uuid", "")
if self.type == "cyclonedx":
parts = uuid.split(":")
if len(parts) == 3 and parts[0] == "urn" and parts[1] == "uuid":
serialNumber = parts[2]
if validate_serialNumber(serialNumber):
# Extract serialNumber (optional in CycloneDX spec)
serialNumber = doc.get("serialNumber", "").lower()
if serialNumber: # Only validate if present
if re.match(
r"^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$",
serialNumber,
):
self.serialNumber = serialNumber
else:
LOGGER.error(
LOGGER.warning( # Downgrade to warning
f"The SBOM file '{self.filename}' has an invalid serial number."
)
return []
else:
LOGGER.error(
f"The SBOM file '{self.filename}' has an invalid serial number."
)
return []
# Do NOT return early; continue parsing components

modules = []
if self.validate and self.filename.endswith(".xml"):
Expand Down Expand Up @@ -281,7 +300,7 @@ def parse_cyclonedx_spdx(self) -> [(str, str, str)]:
# Found at least package and version, save the results
modules.append([vendor, package_name, version])

LOGGER.debug(f"Parsed SBOM {self.filename} {modules}")
LOGGER.debug(f"SBOM Data {self.sbom_data}")
return modules

def parse_swid(self, sbom_file: str) -> list[list[str]]:
Expand Down Expand Up @@ -372,7 +391,7 @@ def decode_purl(self, purl) -> (str | None, str | None, str | None):
- purl (str): Package URL (purl) string.

Returns:
- Tuple[str | None, str | None, str | None]: A tuple containing the vendor (which is always None for purl),
- Tuple[str | None, str | None, str | None]]: A tuple containing the vendor (which is always None for purl),
product, and version information extracted from the purl string, or None if the purl is invalid or incomplete.

"""
Expand Down
151 changes: 123 additions & 28 deletions cve_bin_tool/sbom_manager/sbom_detection.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,51 +2,146 @@
# SPDX-License-Identifier: GPL-3.0-or-later

import json
from typing import Optional

import defusedxml.ElementTree as ET

from cve_bin_tool.validator import validate_cyclonedx, validate_swid
from cve_bin_tool.log import LOGGER
from cve_bin_tool.validator import validate_cyclonedx


def sbom_detection(file_path: str) -> str:
def sbom_detection(file_path: str) -> Optional[str]:
"""
Identifies SBOM type of file based on its format and schema.
Identifies SBOM type with content validation and extension checks.
Returns 'spdx', 'cyclonedx', 'swid', or None if the SBOM type cannot be determined.

Args:
file_path (str): The path to the file.
file_path (str): Path to the SBOM file.

Returns:
str: The detected SBOM type (spdx, cyclonedx, swid) or None.
Optional[str]: The detected SBOM type or None if detection fails.
"""
try:
with open(file_path) as file:
if ".spdx" in file_path:
return "spdx"

elif file_path.endswith(".json"):
data = json.load(file)
if (
"bomFormat" in data
and "specVersion" in data
and data["bomFormat"] == "CycloneDX"
):
# Check for CycloneDX JSON format
if file_path.lower().endswith(".json"):
try:
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
# Check for CycloneDX-specific structure
if isinstance(data, dict):
if (
data.get("bomFormat") == "CycloneDX"
and "components" in data
):
return "cyclonedx"
# Fallback: Check for required fields
if "specVersion" in data.get(
"bom", {}
) and "version" in data.get("bom", {}):
LOGGER.warning(
f"Possible CycloneDX SBOM with non-standard structure: {file_path}"
)
return "cyclonedx"
except (json.JSONDecodeError, UnicodeDecodeError) as e:
LOGGER.debug(f"JSON parsing error for {file_path}: {str(e)}")
pass # Not JSON, continue with other checks

# Check XML-based formats with namespace validation
if file_path.endswith(".xml"):
try:
tree = ET.parse(file_path)
root = tree.getroot()
namespace = (
root.tag.split("}", 1)[0].strip("{") if "}" in root.tag else ""
)

# Check CycloneDX namespace
if "cyclonedx.org" in namespace and validate_cyclonedx(file_path):
Comment thread Fixed
return "cyclonedx"
# Check SWID by root tag and namespace
elif root.tag.endswith("SoftwareIdentity") and "iso/19770" in namespace:
return "swid"
except ET.ParseError as e:
LOGGER.debug(f"XML parsing error for {file_path}: {str(e)}")
return None

# SPDX detection (case-insensitive and path check)
if any(
ext in file_path.lower()
for ext in [".spdx", ".spdx.json", ".spdx.xml", ".spdx.yml", ".spdx.yaml"]
):
return "spdx"

except Exception as e:
LOGGER.error(f"SBOM detection failed for {file_path}: {str(e)}")
return None


def detect_sbom_type_from_content(file_path: str) -> Optional[str]:
"""
Detects SBOM type by analyzing file content without relying on file extensions.
This is a fallback method if the primary detection fails.

Args:
file_path (str): Path to the SBOM file.

Returns:
Optional[str]: The detected SBOM type or None if detection fails.
"""
try:
with open(file_path, "rb") as f:
content = f.read(1024) # Read first 1KB for analysis

else:
return None
# Check for JSON content
if content.startswith(b"{"):
try:
with open(file_path, encoding="utf-8") as f:
data = json.load(f)
if isinstance(data, dict) and data.get("bomFormat") == "CycloneDX":
return "cyclonedx"
elif "SPDXID" in data or "spdxVersion" in data:
return "spdx"
except json.JSONDecodeError:
pass

elif file_path.endswith(".xml"):
# Check for XML content
if content.startswith(b"<?xml"):
try:
tree = ET.parse(file_path)
root = tree.getroot()
root_tag = root.tag.split("}")[-1] if "}" in root.tag else root.tag
if root_tag == "bom" and validate_cyclonedx(file_path):
namespace = (
root.tag.split("}", 1)[0].strip("{") if "}" in root.tag else ""
)

if "cyclonedx.org" in namespace:
Comment thread Fixed
return "cyclonedx"
elif root_tag == "SoftwareIdentity" and validate_swid(file_path):
elif "iso/19770" in namespace:
return "swid"
else:
return None
else:
return None
elif "spdx.org" in namespace:
Comment thread Fixed
return "spdx"
except ET.ParseError:
pass

except Exception as e:
LOGGER.debug(f"Content-based SBOM detection failed for {file_path}: {str(e)}")
return None


def detect_sbom(file_path: str) -> Optional[str]:
"""
Detects SBOM type using both file extension and content-based methods.
This is the main function to be used for SBOM detection.

Args:
file_path (str): Path to the SBOM file.

Returns:
Optional[str]: The detected SBOM type or None if detection fails.
"""
# First, try detection based on file extension and content
sbom_type = sbom_detection(file_path)
if sbom_type:
return sbom_type

except (json.JSONDecodeError, ET.ParseError):
return None
# If primary detection fails, try content-based detection as a fallback
return detect_sbom_type_from_content(file_path)
Loading