-
Notifications
You must be signed in to change notification settings - Fork 52
feat(sbom): merge Maturin components into wheel SBOM #1327
New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
base: main
Are you sure you want to change the base?
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
|
|
@@ -10,13 +10,14 @@ | |
| import json | ||
| import logging | ||
| import pathlib | ||
| import re | ||
| import typing | ||
| from datetime import UTC, datetime | ||
|
|
||
| from packageurl import PackageURL | ||
| from packaging.requirements import Requirement | ||
| from packaging.utils import NormalizedName, canonicalize_name | ||
| from packaging.version import Version | ||
| from packaging.version import InvalidVersion, Version | ||
|
|
||
| if typing.TYPE_CHECKING: | ||
| from . import context | ||
|
|
@@ -181,6 +182,268 @@ def generate_sbom( | |
| return doc | ||
|
|
||
|
|
||
| # CycloneDX hash algorithm names mapped to their SPDX equivalents. | ||
| _CYCLONEDX_HASH_TO_SPDX = { | ||
| "SHA-1": "SHA1", | ||
| "SHA-224": "SHA224", | ||
| "SHA-256": "SHA256", | ||
| "SHA-384": "SHA384", | ||
| "SHA-512": "SHA512", | ||
| "SHA3-224": "SHA3-224", | ||
| "SHA3-256": "SHA3-256", | ||
| "SHA3-384": "SHA3-384", | ||
| "SHA3-512": "SHA3-512", | ||
| "BLAKE2B-256": "BLAKE2b-256", | ||
| "BLAKE2B-384": "BLAKE2b-384", | ||
| "BLAKE2B-512": "BLAKE2b-512", | ||
| } | ||
|
|
||
|
|
||
| def _clean_purl(purl: str) -> str: | ||
| """Drop local ``file://`` download qualifiers, which are build-only paths.""" | ||
| try: | ||
| parsed = PackageURL.from_string(purl) | ||
| except ValueError: | ||
| return purl | ||
| qualifiers = dict(parsed.qualifiers or {}) | ||
| if not qualifiers.get("download_url", "").startswith("file://"): | ||
| return purl | ||
| del qualifiers["download_url"] | ||
| return PackageURL( | ||
| type=parsed.type, | ||
| namespace=parsed.namespace, | ||
| name=parsed.name, | ||
| version=parsed.version, | ||
| qualifiers=qualifiers or None, | ||
| subpath=parsed.subpath, | ||
| ).to_string() | ||
|
|
||
|
|
||
| def _iter_components( | ||
| component: dict[str, typing.Any], | ||
| ) -> typing.Iterator[dict[str, typing.Any]]: | ||
| """Yield a component and all of its nested sub-components.""" | ||
| yield component | ||
| for nested in component.get("components", []): | ||
| yield from _iter_components(nested) | ||
|
|
||
|
|
||
| def _component_key(component: dict[str, typing.Any]) -> str: | ||
| """Return a stable identity used to deduplicate components across files.""" | ||
| purl = component.get("purl") | ||
| if purl: | ||
| return _clean_purl(purl) | ||
| return "\x00".join(component.get(field, "") for field in ("type", "group", "name")) | ||
|
|
||
|
|
||
| def _versions_match(left: str, right: str) -> bool: | ||
| try: | ||
| return Version(left) == Version(right) | ||
| except InvalidVersion: | ||
| return left == right | ||
|
|
||
|
|
||
| def _matches_wheel( | ||
| component: dict[str, typing.Any], | ||
| wheel: dict[str, typing.Any], | ||
| ) -> bool: | ||
| """True if a CycloneDX root is the wheel itself (e.g. an auditwheel root). | ||
|
|
||
| Such a root is folded into ``SPDXRef-wheel`` so its native dependencies are | ||
| not attached to the upstream source. | ||
| """ | ||
| purl = component.get("purl") | ||
| if not purl: | ||
| return False | ||
| try: | ||
| parsed = PackageURL.from_string(purl) | ||
| except ValueError: | ||
| return False | ||
| if parsed.type != "pypi" or not parsed.name or not parsed.version: | ||
| return False | ||
| return canonicalize_name(parsed.name) == canonicalize_name( | ||
| wheel["name"] | ||
| ) and _versions_match(parsed.version, wheel["versionInfo"]) | ||
|
|
||
|
|
||
| def _cyclonedx_license(component: dict[str, typing.Any]) -> str | None: | ||
| """Return the component's declared license as an SPDX expression. | ||
|
|
||
| CycloneDX allows either a single SPDX ``expression`` (already valid SPDX, | ||
| passed through unchanged) or a list of license objects. cargo/maturin always | ||
| emit exactly one entry. A list with multiple entries has undefined AND/OR | ||
| semantics, so we warn and skip it rather than guess a relationship. A named | ||
| (non-SPDX) license has no valid SPDX identifier and is likewise skipped. | ||
| """ | ||
| licenses = component.get("licenses", []) | ||
| if not licenses: | ||
| return None | ||
| if len(licenses) > 1: | ||
| logger.warning( | ||
| "component %s has %d license entries with undefined AND/OR " | ||
| "semantics; skipping license", | ||
| component.get("purl") or component.get("name"), | ||
| len(licenses), | ||
| ) | ||
| return None | ||
| entry = licenses[0] | ||
| expression = entry.get("expression") | ||
| if expression: | ||
| return str(expression) | ||
|
Comment on lines
+290
to
+292
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win Validate CycloneDX expressions before assigning
Use the same validated parser pattern as ♻️ Sketch+from license_expression import ExpressionError, get_spdx_licensing
+
expression = entry.get("expression")
if expression:
- return str(expression)
+ try:
+ get_spdx_licensing().parse(str(expression), validate=True)
+ except ExpressionError:
+ logger.warning(
+ "component %s has an invalid SPDX license expression %r; "
+ "skipping license",
+ component.get("purl") or component.get("name"),
+ expression,
+ )
+ return None
+ return str(expression)🤖 Prompt for AI Agents |
||
| license_info = entry.get("license") | ||
| if isinstance(license_info, dict): | ||
| identifier = license_info.get("id") | ||
| return str(identifier) if identifier else None | ||
| return None | ||
|
|
||
|
|
||
| def _cyclonedx_checksums(component: dict[str, typing.Any]) -> list[dict[str, str]]: | ||
| """Convert a component's hashes into SPDX checksum entries.""" | ||
| checksums = [] | ||
| for entry in component.get("hashes", []): | ||
| algorithm = _CYCLONEDX_HASH_TO_SPDX.get(entry.get("alg", "").upper()) | ||
| content = entry.get("content") | ||
| if algorithm and content: | ||
| checksums.append({"algorithm": algorithm, "checksumValue": content}) | ||
| return checksums | ||
|
Comment on lines
+300
to
+308
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win Validate raw CycloneDX checksum content before copying it. When SBOM generation is enabled, algorithm = _CYCLONEDX_HASH_TO_SPDX.get(entry.get("alg", "").upper())
content = entry.get("content")
- if algorithm and content:
+ if (
+ algorithm
+ and isinstance(content, str)
+ and content
+ and re.fullmatch(r"[0-9a-fA-F]+", content)
+ ):
checksums.append({"algorithm": algorithm, "checksumValue": content})
+ elif algorithm and content:
+ logger.warning(
+ "component %s has invalid %s checksum content; skipping",
+ component.get("purl") or component.get("name"),
+ algorithm,
+ )🤖 Prompt for AI Agents |
||
|
|
||
|
|
||
| def _cyclonedx_package( | ||
| component: dict[str, typing.Any], | ||
| spdx_id: str, | ||
| ) -> dict[str, typing.Any]: | ||
| """Build an SPDX package entry from a CycloneDX component.""" | ||
| purl = component.get("purl") | ||
| purl = _clean_purl(purl) if purl else None | ||
| package: dict[str, typing.Any] = { | ||
| "SPDXID": spdx_id, | ||
| "name": component.get("name") or purl or component.get("bom-ref") or "unknown", | ||
| "versionInfo": component.get("version") or "NOASSERTION", | ||
| "downloadLocation": "NOASSERTION", | ||
| "supplier": "NOASSERTION", | ||
| } | ||
| if purl: | ||
| package["externalRefs"] = [ | ||
| { | ||
| "referenceCategory": "PACKAGE-MANAGER", | ||
| "referenceType": "purl", | ||
| "referenceLocator": purl, | ||
| } | ||
| ] | ||
| checksums = _cyclonedx_checksums(component) | ||
| if checksums: | ||
| package["checksums"] = checksums | ||
| license_expression = _cyclonedx_license(component) | ||
| if license_expression: | ||
| package["licenseDeclared"] = license_expression | ||
| scope = component.get("scope") | ||
| if scope and scope != "required": | ||
| package["comment"] = f"CycloneDX scope: {scope}" | ||
| return package | ||
|
|
||
|
|
||
| def _spdx_id(component: dict[str, typing.Any], used_ids: set[str]) -> str: | ||
| """Build a readable, unique SPDXID from the component name and version. | ||
|
|
||
| SPDXIDs allow only ``A-Za-z0-9.-``; other characters (e.g. cargo's ``_``) | ||
| are replaced with ``-``. Distinct components can share a name and version | ||
| (e.g. a crate and its library target), so a numeric suffix disambiguates | ||
| collisions to keep every SPDXID unique. | ||
| """ | ||
| name = component.get("name") or "unknown" | ||
| version = component.get("version") or "unknown" | ||
| base = re.sub(r"[^A-Za-z0-9.-]", "-", f"SPDXRef-{name}-{version}") | ||
| candidate = base | ||
| suffix = 1 | ||
| while candidate in used_ids: | ||
| suffix += 1 | ||
| candidate = f"{base}-{suffix}" | ||
| used_ids.add(candidate) | ||
| return candidate | ||
|
|
||
|
|
||
| def _iter_all_components( | ||
| cyclonedx: dict[str, typing.Any], | ||
| ) -> typing.Iterator[tuple[dict[str, typing.Any], bool]]: | ||
| """Yield every ``(component, is_root)`` pair in a CycloneDX document.""" | ||
| root = cyclonedx.get("metadata", {}).get("component") | ||
| if root: | ||
| for component in _iter_components(root): | ||
| yield component, component is root | ||
| for component in cyclonedx.get("components", []): | ||
| for nested in _iter_components(component): | ||
| yield nested, False | ||
|
|
||
|
|
||
| def merge_cyclonedx_sboms( | ||
| *, | ||
| sbom: dict[str, typing.Any], | ||
| sboms_dir: pathlib.Path, | ||
| ) -> None: | ||
| """Merge Maturin CycloneDX components into a Fromager SPDX document. | ||
|
|
||
| Each non-excluded component becomes an SPDX package linked to the wheel with | ||
| ``CONTAINS``. A CycloneDX root matching the wheel is folded into | ||
| ``SPDXRef-wheel``. Local ``file://`` download qualifiers are stripped from | ||
| PURLs. The CycloneDX dependency graph is not copied and the original files | ||
| are left in place. | ||
| """ | ||
| if not sboms_dir.is_dir(): | ||
| return | ||
|
|
||
| wheel = next( | ||
| (p for p in sbom["packages"] if p.get("SPDXID") == "SPDXRef-wheel"), None | ||
| ) | ||
| packages = sbom["packages"] | ||
| relationships = sbom["relationships"] | ||
| used_ids = {p["SPDXID"] for p in packages} | ||
| key_to_id: dict[str, str] = {} | ||
| contained: set[str] = set() | ||
| merged_files: list[str] = [] | ||
|
|
||
| for sbom_path in sorted(sboms_dir.iterdir()): | ||
| if not sbom_path.is_file() or sbom_path.name == SBOM_FILENAME: | ||
| continue | ||
| try: | ||
| cyclonedx = json.loads(sbom_path.read_text(encoding="utf-8")) | ||
| except (OSError, json.JSONDecodeError) as err: | ||
| logger.warning("could not read SBOM file %s: %s", sbom_path, err) | ||
| continue | ||
| if not isinstance(cyclonedx, dict) or cyclonedx.get("bomFormat") != "CycloneDX": | ||
| continue | ||
| merged_files.append(sbom_path.name) | ||
|
|
||
| for component, is_root in _iter_all_components(cyclonedx): | ||
| if component.get("scope") == "excluded": | ||
| continue | ||
|
|
||
| if is_root and wheel is not None and _matches_wheel(component, wheel): | ||
| spdx_id = "SPDXRef-wheel" | ||
| else: | ||
| key = _component_key(component) | ||
| spdx_id = key_to_id.get(key, "") | ||
| if not spdx_id: | ||
| spdx_id = _spdx_id(component, used_ids) | ||
| key_to_id[key] = spdx_id | ||
| packages.append(_cyclonedx_package(component, spdx_id)) | ||
|
|
||
| if spdx_id != "SPDXRef-wheel" and spdx_id not in contained: | ||
| contained.add(spdx_id) | ||
| relationships.append( | ||
| { | ||
| "spdxElementId": "SPDXRef-wheel", | ||
| "relationshipType": "CONTAINS", | ||
| "relatedSpdxElement": spdx_id, | ||
| } | ||
| ) | ||
|
|
||
| if merged_files: | ||
| sbom["comment"] = ( | ||
| "Includes components merged from CycloneDX SBOM(s): " | ||
| + ", ".join(sorted(merged_files)) | ||
| ) | ||
|
|
||
|
|
||
| def write_sbom( | ||
| *, | ||
| sbom: dict[str, typing.Any], | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick win
🔎 Supported by static analysis
🏁 Script executed:
Repository: python-wheel-build/fromager
Length of output: 344
🏁 Script executed:
Repository: python-wheel-build/fromager
Length of output: 21293
🤖 get_repo_knowledge executed:
get_repo_knowledge python-wheel-build/fromager /tmp/coderabbit-repo-knowledge/python-wheel-build-fromager-63bbc311/learningsLength of output: 4488
🏁 Script executed:
Repository: python-wheel-build/fromager
Length of output: 15303
🏁 Script executed:
Repository: python-wheel-build/fromager
Length of output: 1791
Remove the
SHA3-224mapping._cyclonedx_checksumsemits mapped hashes as SPDX package checksums. SPDX 2.3 and the boundspdx_toolsenum do not defineSHA3-224, so strict validation can reject documents containing this hash. Removing the entry skips it.🔧 Proposed fix
- "SHA3-224": "SHA3-224", "SHA3-256": "SHA3-256",📝 Committable suggestion
🤖 Prompt for AI Agents