# Copyright Kevin Deldycke <[email protected]> and contributors.
#
# This program is Free Software; you can redistribute it and/or
# modify it under the terms of the GNU General Public License
# as published by the Free Software Foundation; either version 2
# of the License, or (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program; if not, write to the Free Software
# Foundation, Inc., 59 Temple Place - Suite 330, Boston, MA 02111-1307, USA.
"""Format-agnostic SBOM base class and export-format enum.
Kept deliberately free of SPDX or CycloneDX dependencies: instantiating
{class}`SBOM` directly is meaningless, but importing the symbols here
is safe even when the optional `[sbom-offline]` extra is not installed.
"""
from __future__ import annotations
import logging
import sys
from importlib.metadata import PackageNotFoundError, version
if sys.version_info >= (3, 11):
from enum import StrEnum
else:
from backports.strenum import StrEnum
TYPE_CHECKING = False
if TYPE_CHECKING:
from collections.abc import Iterator
from pathlib import Path
from ..package import PackageMetadata
from .vulnerabilities import Vulnerability
[docs]
def writer_unavailable_reason(distribution: str, error: Exception) -> str:
"""Explain why an SBOM writer library could not be imported.
Tells apart the library being absent from it being installed but older
than what the renderer imports. The two need opposite remedies, and the
install-the-extra advice is actively misleading for the second: a
distribution shipping an older release satisfies the extra already, so the
reader installs it again, sees no change, and is left with a symbol name
and no idea which package owns it. NixOS hits exactly this, carrying a
`cyclonedx-python-lib` below the floor `[sbom-offline]` declares.
The floor itself is deliberately not repeated here: `error` already names
the symbol that is missing, which is more precise than a version number
and cannot drift away from `pyproject.toml`.
"""
try:
installed = version(distribution)
except PackageNotFoundError:
return (
f"{distribution} is not installed. "
f"Install it with: pip install meta-package-manager[sbom-offline]"
)
return (
f"{distribution} {installed} is installed but too old: {error}. "
f"Upgrade it with: pip install --upgrade meta-package-manager[sbom-offline]"
)
[docs]
class SBOM:
"""Utilities shared by all SBOM classes.
```{seealso}
Anchore's [Syft](https://github.com/anchore/syft) and Microsoft's
[sbom-tool](https://github.com/microsoft/sbom-tool) are mature SPDX
and CycloneDX emitters, useful references for field-population
conventions. Both inventory packages by parsing on-disk databases and
lockfiles, whereas `mpm` queries the live managers directly.
```
"""
def __init__(
self,
export_format: ExportFormat = ExportFormat.JSON,
) -> None:
"""Defaults to JSON export format."""
logging.debug(f"Set export format to {export_format}")
self.export_format = export_format
# `manager_id -> count` of unique packages the renderer admitted
# into the document. Populated by {meth}`_track_addition` so
# subclasses' format-specific dedup is reflected here.
self.packages_per_manager: dict[str, int] = {}
# `manager_id -> count` of admitted packages whose metadata was
# non-empty (i.e. the manager's extractor produced something).
self.enriched_per_manager: dict[str, int] = {}
# Keys used to dedup `_track_addition` calls across subclasses
# that may invoke it more than once per (manager, package) pair.
self._tracked_additions: set[tuple[str, str]] = set()
# `purl string -> vulnerabilities` attached post-hoc by the
# network layer (`mpm --network sbom`). Distinct from
# PackageMetadata, which the local extractor produces: this is
# data fetched after the fact from OSV and bound to the document
# via {meth}`attach_vulnerabilities`. Renderers consume it in
# their `finalize` override.
self.vulnerabilities_by_purl: dict[str, tuple[Vulnerability, ...]] = {}
# `alias purl -> primary purls`, for packages a manager identifies
# through more than one coordinate system. Fed from
# `PackageMetadata.extra_purls` by {meth}`register_purl_aliases`.
# A list rather than a set: two packages can share one upstream
# coordinate, and the emission order must stay deterministic.
self.purl_aliases: dict[str, list[str]] = {}
# Every purl the inventory pass admitted as a package of its own,
# which is what tells a real coordinate apart from an alias that
# only points at one. Also fed by {meth}`register_purl_aliases`.
self.primary_purls: set[str] = set()
[docs]
def register_purl_aliases(
self,
primary_purl: str,
metadata: PackageMetadata,
) -> None:
"""Index a package's `extra_purls` as aliases of its primary purl.
An alias is a second coordinate for the same installed package, so
an advisory found under it belongs to the package the inventory
pass added. Homebrew is the motivating case: a formula's own
`pkg:brew/…` coordinate is unknown to every advisory database,
while the upstream registry coordinate it records alongside is not.
Aliases join {meth}`all_purls`, so the network layer queries them
too, and {meth}`resolve_purl_targets` maps the answers back.
Called once per package, whatever its metadata holds, so it also
registers the primary purl itself.
"""
self.primary_purls.add(primary_purl)
for extra in metadata.extra_purls:
extra_str = extra.to_string()
if extra_str == primary_purl:
continue
owners = self.purl_aliases.setdefault(extra_str, [])
if primary_purl not in owners:
owners.append(primary_purl)
[docs]
def resolve_purl_targets(self, purl_str: str) -> tuple[str, ...]:
"""Return every package purl an advisory key designates.
The key itself comes first, covering the ordinary case where the
scan queried a package's own purl. Its alias owners follow, so a
coordinate that is *both* one package's own purl and another's
alias reaches both: `pkg:pypi/ty@…` names the pip package
directly and the `ty` formula built from it, and the advisory
belongs to each.
Callers index each target against their own package map and skip
what does not resolve, which is what filters out the key when it
names no package of its own.
"""
return (purl_str, *self.purl_aliases.get(purl_str, ()))
[docs]
def all_purls(self) -> Iterator[str]:
"""Yield every package purl present in the document.
Powers the vulnerability scan: the network layer queries OSV once
with the full purl set rather than once per package. Subclasses
implement this against their own component index, and add the
aliases {meth}`register_purl_aliases` collected.
"""
raise NotImplementedError
[docs]
def attach_vulnerabilities(
self,
vulnerabilities: dict[str, tuple[Vulnerability, ...]],
) -> None:
"""Bind cross-package vulnerability data to the document.
Called by the CLI between the per-package `add_package` loop and
`finalize`, only in `--network` mode. Renderers read the
stored data in their `finalize` override and project it into the
format-native vulnerability surface (CycloneDX `vulnerabilities`
array, SPDX security `externalRefs`).
"""
self.vulnerabilities_by_purl.update(vulnerabilities)
def _track_addition(
self,
manager_id: str,
package_id: str,
metadata: PackageMetadata | None,
) -> None:
"""Record that one package entered the document.
Called by {meth}`add_package` subclass implementations after
their own dedup check so the renderer-level counters reflect
what actually got serialized, not the number of inbound calls.
Idempotent on `(manager_id, package_id)` to stay robust against
future refactors that might double-call.
"""
key = (manager_id, package_id)
if key in self._tracked_additions:
return
self._tracked_additions.add(key)
self.packages_per_manager[manager_id] = (
self.packages_per_manager.get(manager_id, 0) + 1
)
if metadata is not None and not metadata.is_empty():
self.enriched_per_manager[manager_id] = (
self.enriched_per_manager.get(manager_id, 0) + 1
)
[docs]
def stats(self) -> dict[str, object]:
"""Return a summary of what landed in the document.
Format-agnostic counters live in the base implementation; SPDX and
CycloneDX subclasses extend the returned dict with their own
merged-documents, dependency-graph, and any other format-specific
counts. Surfaced by the CLI as a post-run INFO-level summary and
usable by tests or programmatic consumers without re-parsing the
rendered document.
"""
# Count unique advisories and the packages they affect. The same
# advisory can affect several packages, so the vulnerability total
# is over distinct ids, not over the per-purl lists. A finding is
# counted against the packages it reaches rather than the
# coordinate it was found under, which are not the same thing once
# an alias is in play: an alias names no package of its own, and
# can name more than one.
affected_purls = {
target
for purl_str, vulns in self.vulnerabilities_by_purl.items()
if vulns
for target in self.resolve_purl_targets(purl_str)
if target in self.primary_purls
}
unique_vuln_ids = {
vuln.id for vulns in self.vulnerabilities_by_purl.values() for vuln in vulns
}
return {
"packages_total": sum(self.packages_per_manager.values()),
"packages_per_manager": dict(self.packages_per_manager),
"enriched_per_manager": dict(self.enriched_per_manager),
"vulnerabilities_total": len(unique_vuln_ids),
"vulnerable_packages": len(affected_purls),
}
[docs]
def finalize(self) -> None:
"""Resolve any deferred state before `export()`.
Some constructs cannot be emitted at `add_package()` time
because they reference packages that may not have been added yet:
a Homebrew formula's runtime dependency on another formula listed
later in the scan, for example. Subclasses queue those during
`add_package` and flush them here. The base implementation is a
no-op so subclasses can rely on it being called exactly once.
"""