feat!(ISV-6786): consume new data structure for generating image SBOMs

Author: bclindner

docs/sboms/oci_image.md

@@ -44,7 +44,6 @@ mobster --verbose  generate oci-image \
 - `--base-image-digest-file` -- points to a file with digests for images used in Dockerfile.
   if omitted, the references will be fetched via `oras`. The expected format of the file is
   `<registry>/<repository>:<tag> <registry>/<repository>:<tag>@sha256:<digest>`
-- `--dockerfile-target` -- if a build target was used for multi-stage build, use this argument to specify the build target
 - `--additional-base-images` -- optionally add references to other build images outside the parsed Dockerfile.
   expects the format `<registry>/<repository>:<tag>@sha256:<digest value>`
 - `--contextualize` -- Allows SBOM contextualization (see [Contextual SBOM](#contextual-sbom))

poetry.lock

@@ -36,6 +36,7 @@ dependencies = [
     "aiofiles (>=24.1.0,<25.0.0)",
     "httpx (>=0.28.1,<0.29.0)",
     "aioboto3 (>=15.2.0,<15.3.0)",
+    "pyyaml (>=6.0.3,<7.0.0)",
 ]
 
 [project.urls]

pyproject.toml

@@ -130,31 +130,9 @@ def validated_additional_reference(value: str) -> str:
         help="Image digest for the OCI image in the format sha256:<digest>",
     )
     oci_image_parser.add_argument(
-        "--parsed-dockerfile-path",
+        "--metadata-path",
         type=Path,
-        help="Path to the parsed Dockerfile file",
-    )
-    oci_image_parser.add_argument(
-        "--base-image-digest-file",
-        type=Path,
-        help="Path to the file containing references "
-        "to images in the Dockerfile and their digests. "
-        "Expected format: "
-        "`<registry>/<repository>:<tag> <registry>/<repository>:<tag>@sha256:<digest>`",
-    )
-    oci_image_parser.add_argument(
-        "--dockerfile-target",
-        type=str,
-        help="The name of the build target from the Dockerfile",
-        default=None,
-    )
-    oci_image_parser.add_argument(
-        "--additional-base-image",
-        type=validated_additional_reference,
-        action="append",
-        default=[],
-        help="Base (builder) image to add, can be specified multiple times. "
-        "Expects the format <registry>/<repository>:<tag>@sha256:<digest value>",
+        help="Path to a metadata file generated by sbomgen.",
     )
     oci_image_parser.add_argument(
         "--arch",

src/mobster/cli.py

@@ -10,20 +10,20 @@
 from typing import Any
 
 from cyclonedx.exception import CycloneDxException
+from mobster.cmd.generate.oci_image.sbomgen import SBOMMetadata
 from spdx_tools.spdx.jsonschema.document_converter import DocumentConverter
 from spdx_tools.spdx.model.document import Document
 from spdx_tools.spdx.validation.document_validator import validate_full_spdx_document
 from spdx_tools.spdx.writer.write_utils import convert
+import yaml
 
 import mobster.utils
 from mobster import syft
 from mobster.cmd.generate.base import GenerateCommandWithOutputTypeSelector
 from mobster.cmd.generate.oci_image.add_image import extend_sbom_with_image_reference
 from mobster.cmd.generate.oci_image.base_images_dockerfile import (
-    extend_sbom_with_base_images_from_dockerfile,
-    get_base_images_refs_from_dockerfile,
+    extend_sbom_with_base_images,
     get_digest_for_image_ref,
-    get_image_objects_from_file,
 )
 from mobster.cmd.generate.oci_image.contextual_sbom.builder import (
     BuilderContextualizationError,
@@ -99,6 +99,15 @@ async def _load_and_filter_hermeto_sbom(self) -> dict[str, Any]:
         arch = self.cli_args.arch or mobster.utils.identify_arch()
         return filter_hermeto_sbom_by_arch(hermeto_sbom, arch)
 
+    def _load_metadata(self) -> None:
+        """
+        Load a metadata file from the --metadata-path argument into
+        self._metadata.
+        """
+        with open(self.cli_args.metadata_path) as metadata_file:
+            raw_metadata = yaml.load(metadata_file, yaml.Loader)
+            self._metadata = SBOMMetadata.from_dict(raw_metadata)
+
     async def _handle_bom_inputs(
         self,
     ) -> dict[str, Any]:
@@ -113,13 +122,19 @@ async def _handle_bom_inputs(
             self.cli_args.from_hermeto is None
             and self.cli_args.from_syft is None
             and self.cli_args.image_pullspec is None
+            and self.cli_args.metadata_path is None
         ):
             raise ArgumentError(
                 None,
-                "At least one of --from-syft, --from-hermeto or --image-pullspec"
-                " must be provided",
+                "At least one of --from-syft, --from-hermeto, --image-pullspec, "
+                "or --metadata-path must be provided",
             )
 
+        if self.cli_args.metadata_path is not None:
+            self._load_metadata()
+            # if we don't have an sbom provided to us, use syft to generate it
+            if self.cli_args.from_syft is None and self.cli_args.from_hermeto is None:
+                return await syft.scan_image(self._metadata.image.pullspec)
         if self.cli_args.from_syft is not None:
             # Merging Syft & Hermeto SBOMs
             if len(self.cli_args.from_syft) > 1 or self.cli_args.from_hermeto:
@@ -228,7 +243,7 @@ async def _assess_and_dispatch_contextual_workflow(
         (non-modified) SBOM is furtherly processed by mobster.
         Args:
             component_sbom_doc: The component SBOM created for this image.
-            base_images_refs: List of references from the parsed Dockerfile.
+            base_images_refs: List of references from the build.
             image_arch: CPU architecture of this image.
 
         Returns:
@@ -264,11 +279,12 @@ async def execute(self) -> Any:
         """
         LOGGER.debug("Generating SBOM document for OCI image")
 
+        # Get/merge the raw SBOM
         merged_sbom_dict = await self._handle_bom_inputs()
         sbom: Document | CycloneDX1BomWrapper
-        image_arch = identify_arch()
+        image_arch = self.cli_args.arch or mobster.utils.identify_arch()
 
-        # Parsing into objects
+        # Parse into objects
         if merged_sbom_dict.get("bomFormat") == "CycloneDX":
             if self.cli_args.contextualize:
                 raise ArgumentError(
@@ -280,9 +296,32 @@ async def execute(self) -> Any:
         else:
             raise ValueError("Unknown SBOM Format!")
 
-        # Extending with image reference
-        if self.cli_args.image_pullspec:
-            image_arch = self.cli_args.arch or mobster.utils.identify_arch()
+        base_images_refs = []
+        base_images_map: dict[str, Image] = {}
+
+        # Extend with image reference
+        if self.cli_args.metadata_path:
+            image = Image.from_image_index_url_and_digest(
+                self._metadata.image.pullspec,
+                self._metadata.image.digest,
+                arch=image_arch,
+            )
+            await extend_sbom_with_image_reference(sbom, image, False)
+            for base_image_data in self._metadata.base_images:
+                base_image = Image.from_image_index_url_and_digest(
+                    base_image_data.pullspec,
+                    base_image_data.digest,
+                )
+                base_images_refs.append(base_image_data.pullspec)
+                base_images_map[base_image_data.pullspec] = base_image
+            await extend_sbom_with_base_images(sbom, base_images_refs, base_images_map)
+            for extra_image_data in self._metadata.extra_images:
+                extra_image = Image.from_image_index_url_and_digest(
+                    extra_image_data.pullspec,
+                    extra_image_data.digest,
+                )
+                await extend_sbom_with_image_reference(sbom, extra_image, True)
+        elif self.cli_args.image_pullspec:
             if not self.cli_args.image_digest:
                 LOGGER.info(
                     "Provided pullspec but not digest."
@@ -308,37 +347,6 @@ async def execute(self) -> Any:
                 "Provided image digest but no pullspec. The digest value is ignored."
             )
 
-        base_images_refs = []
-        base_images_map: dict[str, Image] = {}
-
-        # Extending with base images references from a dockerfile
-        if self.cli_args.parsed_dockerfile_path:
-            with open(
-                self.cli_args.parsed_dockerfile_path, encoding="utf-8"
-            ) as parsed_dockerfile_io:
-                parsed_dockerfile = json.load(parsed_dockerfile_io)
-
-            base_images_refs = await get_base_images_refs_from_dockerfile(
-                parsed_dockerfile, self.cli_args.dockerfile_target
-            )
-
-            if self.cli_args.base_image_digest_file:
-                LOGGER.debug(
-                    "Supplied pre-parsed image digest file, will operate offline."
-                )
-                base_images_map = await get_image_objects_from_file(
-                    self.cli_args.base_image_digest_file
-                )
-            await extend_sbom_with_base_images_from_dockerfile(
-                sbom, base_images_refs, base_images_map
-            )
-
-        # Extending with additional base images
-        for image_ref in self.cli_args.additional_base_image:
-            image_object = Image.from_oci_artifact_reference(image_ref)
-            await extend_sbom_with_image_reference(
-                sbom, image_object, is_builder_image=True
-            )
         with log_elapsed("Contextual workflow", logging.INFO):
             contextual_sbom = await self._assess_and_dispatch_contextual_workflow(
                 sbom, base_images_refs, base_images_map, image_arch

src/mobster/cmd/generate/oci_image/__init__.py

@@ -1,4 +1,4 @@
-"""Module for augmenting the oci-image SBOM with information from a parsed Dockerfile"""
+"""Module for augmenting the oci-image SBOM with information from various sources"""
 
 import json
 import logging
@@ -27,74 +27,6 @@
 LOGGER = logging.getLogger(__name__)
 
 
-async def get_base_images_refs_from_dockerfile(
-    parsed_dockerfile: dict[str, Any], target_stage: str | None = None
-) -> list[str | None]:
-    """
-    Reads the base images from provided parsed dockerfile, does not include
-    stages after the target of the build. So the last image returned is
-    the parent image used.
-
-    Args:
-        parsed_dockerfile (dict[str, Any]): Contents of the parsed dockerfile
-        target_stage (str): The target stage for the build
-    Returns:
-        list[str | None]: List of base images used during build as extracted
-                          from the dockerfile in the order they were used.
-                          `FROM SCRATCH` is identified as `None`.
-
-    Example:
-    If the Dockerfile looks like
-    FROM registry.access.redhat.com/ubi8/ubi:latest as builder
-    ...
-    FROM builder
-    ...
-
-    Then the relevant part of parsed_dockerfile look like
-    {
-        "Stages": [
-            {
-                "BaseName": "registry.access.redhat.com/ubi8/ubi:latest",
-                "As": "builder",
-                "From": {"Image": "registry.access.redhat.com/ubi8/ubi:latest"},
-            },
-            {
-                "BaseName": "builder",
-                "From": {"Stage": {"Named": "builder", "Index": 0}},
-            },
-        ]
-    },
-    """
-    base_images_pullspecs: list[str | None] = []
-    for stage in parsed_dockerfile.get("Stages", []):
-        is_actually_image = True
-
-        from_field = stage.get("From", {})
-        # Ignore scratch image as well as
-        # references to previous stages
-        if "Stage" in from_field:
-            is_actually_image = False
-        if from_field.get("Scratch"):
-            # It is an empty image
-            base_images_pullspecs.append(None)
-            is_actually_image = False
-        base_name: str = stage.get("BaseName")
-        if is_actually_image and base_name and not base_name.startswith("oci-archive:"):
-            # flatpak archives are not real base images. So we skip them
-            base_images_pullspecs.append(base_name.strip("'\""))
-
-        # Don't include images after the target used for build
-        alias = stage.get("As")
-        if target_stage and alias and alias == target_stage:
-            # The `AS` keyword of this stage matches the target
-            break
-        if target_stage and not alias and base_name == target_stage:
-            # This stage does not use the `AS` keyword,
-            # the pull-spec matches the target
-            break
-    return base_images_pullspecs
-
-
 async def get_digest_for_image_ref(image_ref: str, arch: Any = None) -> str | None:
     """
     Fetches the digest of a pullspec using oras.
@@ -142,27 +74,6 @@ def get_base_images_digests_lines(base_images_digests: Path) -> list[str]:
         return list(input_file_stream)
 
 
-async def get_image_objects_from_file(base_images_digests: Path) -> dict[str, Image]:
-    """
-    Parses the base image digest file into a dictionary of
-    image references present in a Dockerfile and Image
-    objects.
-    Args:
-        base_images_digests (Path): File containing the digests of images.
-            expects the format <image_ref> <name>:<tag>@sha256:<digest>
-
-    Returns:
-        dict[str, Image]: Mapping of the references to Image objects
-    """
-    base_images_mapping = {}
-    for line in get_base_images_digests_lines(base_images_digests):
-        line = line.strip()
-        image_ref, image_full_reference = re.split(r"\s+", line)
-        image_obj = Image.from_oci_artifact_reference(image_full_reference.strip("'\""))
-        base_images_mapping[image_ref.strip("'\"")] = image_obj
-    return base_images_mapping
-
-
 async def get_objects_for_base_images(
     base_images_refs: list[str | None],
 ) -> dict[str, Image]:
@@ -221,8 +132,8 @@ async def _get_images_and_their_annotations(
         if not image_obj:
             LOGGER.warning(
                 "Cannot get information about base image "
-                "%s mentioned in the Dockerfile! THIS MEANS "
-                "THE PRODUCED SBOM WILL BE INCOMPLETE!",
+                "%s! THIS MEANS THE PRODUCED SBOM WILL BE"
+                "INCOMPLETE!",
                 image_ref,
             )
             continue
@@ -409,7 +320,7 @@ async def _extend_cdx_with_base_images(
     )
 
 
-async def extend_sbom_with_base_images_from_dockerfile(
+async def extend_sbom_with_base_images(
     sbom: CycloneDX1BomWrapper | Document,
     base_images_refs: list[str | None],
     base_images_objects: dict[str, Image] | None = None,