Merge pull request #626 from open-component-model/zkdev-mv-func

Mv function from cc-utils

Author: zkdev

cli/_bdba.py

@@ -7,7 +7,6 @@
 import tabulate
 
 import cnudie.access
-import oci
 import ocm
 import ocm.iter
 import tarutil
@@ -155,7 +154,7 @@ def iter_resource_scans() -> collections.abc.Generator[odg.model.ArtefactMetadat
             access = resource_node.resource.access
 
             if access.type is ocm.AccessType.OCI_REGISTRY:
-                content_iterator = oci.image_layers_as_tarfile_generator(
+                content_iterator = ocm_util.image_layers_as_tarfile_generator(
                     image_reference=access.imageReference,
                     oci_client=oci_client,
                     include_config_blob=False,

ocm_util.py

@@ -1,5 +1,6 @@
 import collections.abc
 import logging
+import tarfile
 
 import cnudie.access
 import cnudie.retrieve_async
@@ -183,7 +184,7 @@ def iter_content_for_resource_node(
     access = resource_node.resource.access
 
     if access.type is ocm.AccessType.OCI_REGISTRY:
-        return oci.image_layers_as_tarfile_generator(
+        return image_layers_as_tarfile_generator(
             image_reference=access.imageReference,
             oci_client=oci_client,
             include_config_blob=False,
@@ -225,3 +226,70 @@ def iter_content_for_resource_node(
 
     else:
         raise RuntimeError(f'Unsupported access type: {access.type}')
+
+
+def image_layers_as_tarfile_generator(
+    image_reference: str,
+    oci_client: oci.client.Client,
+    chunk_size=tarfile.RECORDSIZE,
+    include_config_blob=True,
+    fallback_to_first_subimage_if_index=False,
+) -> collections.abc.Generator[bytes, None, None]:
+    '''
+    returns a generator yielding a tar-archive with the passed oci-image's layer-blobs as
+    members. This is somewhat similar to the result of a `docker save` with the notable difference
+    that the cfg-blob is discarded.
+    This function is useful to e.g. upload file system contents of an oci-container-image to some
+    scanning-tool (provided it supports the extraction of tar-archives)
+    If include_config_blob is set to False the config blob will be ignored.
+
+    If fallback_to_first_subimage_if_index is set to True, in case of oci-image-manifest-list the
+    first sub-manifest is taken.
+    '''
+    manifest = oci_client.manifest(
+        image_reference=image_reference,
+        accept=oci.model.MimeTypes.prefer_multiarch,
+    )
+
+    image_reference = oci.model.OciImageReference.to_image_ref(image_reference)
+
+    if fallback_to_first_subimage_if_index and isinstance(manifest, oci.model.OciImageManifestList):
+        logger.warn(
+            f'image-index handling not fully implemented - will only scan first image, '
+            f'{image_reference=}, {manifest.mediaType=}'
+        )
+        manifest_ref = manifest.manifests[0]
+        manifest = oci_client.manifest(
+            image_reference=f'{image_reference.ref_without_tag}@{manifest_ref.digest}',
+        )
+
+    blob_refs = manifest.blobs() if include_config_blob else manifest.layers
+
+    if not include_config_blob:
+        logger.debug('skipping config blob')
+
+    def resolve_blob(
+        blob_ref: oci.model.OciBlobRef,
+        image_reference: str,
+    ) -> ioutil.BlobDescriptor:
+        content = oci_client.blob(
+            image_reference=image_reference,
+            digest=blob_ref.digest,
+            stream=True,
+        ).iter_content(chunk_size=chunk_size)
+
+        return ioutil.BlobDescriptor(
+            content=content,
+            size=blob_ref.size,
+            name=f'{blob_ref.digest}.tar',
+        )
+
+    return tarutil.concat_blobs_as_tarstream(
+        blobs=(
+            resolve_blob(
+                blob_ref=blob_ref,
+                image_reference=image_reference,
+            )
+            for blob_ref in blob_refs
+        ),
+    )