feat(RELEASE-2460): add disk-image support to artifact helpers

- compress_artifacts: copy disk images as-is; fix multi-file
  per-arch scenario (e.g. ISO + QCOW2 sharing one os+arch)
- push_unsigned: pass disk images through without unpacking,
  handling both files[] and staged.files[] sources
- push_artifacts: inject staged.files[] into files[] for CGW
  when disk-image component has no files[] entries
- extract_artifacts: fail early when disk-image component
  declares a non-linux OS entry, preventing signing failures
- all helpers: detect disk-image via contentType at both
  component level and contentGateway.contentType

Assisted-by: Cursor AI
Signed-off-by: Scott Wickersham <swickers@redhat.com>

Author: swickersh

scripts/python/helpers/compress_artifacts.py

@@ -5,7 +5,8 @@
 * Pulls signed macOS and Windows OCI artifacts from Quay into a ``signed/`` directory.
 * Restores supplementary files (readme, license, changelog) that were held during signing.
 * Compresses each file entry into the final deliverable format:
-  - macOS / Linux → ``.tar.gz`` (from ``os/arch/`` directory)
+  - macOS / Linux (non-disk-image) → ``.tar.gz`` (from ``os/arch/`` directory)
+  - Linux disk images (``.qcow2``, ``.iso``) → copied as-is to ``ready_for_distribution/``
   - Windows → ``.zip`` (from ``os/arch/`` directory, extension corrected from
     ``.tar.gz``/``.tar``)
 * Updates ``SNAPSHOT_JSON`` to reflect corrected Windows filenames in ``files[]``.
@@ -39,6 +40,23 @@
 
 PROG = "compress_artifacts.py"
 
+# Unambiguous disk-image file suffixes (simple and compound). Files matching
+# these are copied as-is to ready_for_distribution without being wrapped in a
+# tar archive, even when the component does not carry contentType: disk-image.
+# NOTE: .tar.gz is intentionally excluded — it is ambiguous between binary
+# archives and disk images (e.g. GCP images packaged as tarballs). Use
+# contentType: disk-image on the component to handle those cases.
+_DISK_IMAGE_SUFFIXES: frozenset[str] = frozenset(
+    {".qcow2", ".iso", ".iso.gz", ".raw.gz", ".vhd.gz"}
+)
+
+
+def _is_disk_image_file(filename: str) -> bool:
+    """Return True if *filename* has an unambiguous disk-image file suffix."""
+    lower = filename.lower()
+    return any(lower.endswith(ext) for ext in _DISK_IMAGE_SUFFIXES)
+
+
 QUAY_SECRET_MOUNT = Path(os.environ.get("QUAY_SECRET_MOUNT", "/mnt/quaySecret"))
 CONTENT_DIR = Path(os.environ.get("CONTENT_DIR", "/shared/artifacts"))
 SHARED_DIR = Path(os.environ.get("SHARED_DIR", "/shared"))
@@ -103,19 +121,25 @@ def _windows_filename(source_filename: str) -> str:
 
 
 def _compress_file_entry(
-    entry: dict, array_name: str, component_dir: Path, ready_dir: Path
+    entry: dict,
+    array_name: str,
+    component_dir: Path,
+    ready_dir: Path,
+    *,
+    is_disk_image_component: bool = False,
 ) -> str:
     """Compress one file entry into ready_dir and return the (possibly normalized) source path.
 
     For macOS and Linux entries the source path is returned unchanged. For Windows entries
     the archive is created as a ``.zip`` instead of ``.tar.gz``/``.tar``, and the returned
     source path reflects the corrected filename so the snapshot can be updated accordingly.
 
-    Raises RuntimeError on failure (missing source, unknown OS, or empty arch directory).
+    Files are copied directly to ``ready_dir`` (without archiving) when either:
+    - *is_disk_image_component* is True (set when contentType: disk-image), or
+    - the filename has an unambiguous disk-image suffix (.qcow2, .iso, .iso.gz,
+      .raw.gz, .vhd.gz).
 
-    Note: all files are currently compressed regardless of type. ISOs should be
-    passed through as-is rather than wrapped in a tarball — this will need to be
-    addressed before ISO delivery is supported.
+    Raises RuntimeError on failure (missing source, unknown OS, or empty arch directory).
     """
     source = entry.get("source")
     if not source:
@@ -139,12 +163,23 @@ def _compress_file_entry(
 
     # macOS and Linux follow the Unix convention of tar.gz archives; Windows uses zip
     # because that is the standard expected by Windows users and Developer Portal tooling.
+    # Disk images are an exception: they are delivered as-is without any archiving.
     if os_name in ("darwin", "linux"):
         out_path = ready_dir / source_filename
-        with tarfile.open(str(out_path), "w:gz") as tf:
-            for item in sorted(arch_dir.rglob("*")):
-                if item.is_file():
-                    tf.add(str(item), arcname=str(item.relative_to(arch_dir)))
+        if is_disk_image_component or _is_disk_image_file(source_filename):
+            # Use the known filename directly — multiple disk images may share
+            # the same arch directory, so scanning the whole dir is incorrect.
+            src_file = arch_dir / source_filename
+            if not src_file.is_file():
+                raise RuntimeError(
+                    f"Disk image file '{source_filename}' not found in {arch_dir}"
+                )
+            shutil.copy2(str(src_file), str(out_path))
+        else:
+            with tarfile.open(str(out_path), "w:gz") as tf:
+                for item in sorted(arch_dir.rglob("*")):
+                    if item.is_file():
+                        tf.add(str(item), arcname=str(item.relative_to(arch_dir)))
         logger.info("  Created (%s): %s", array_name, source_filename)
         return source
 
@@ -174,13 +209,20 @@ def compress_component(component: dict, snapshot: dict) -> dict:
     files_entries = list(component.get("files") or [])
     staged_entries = list((component.get("staged") or {}).get("files") or [])
 
+    is_disk_image = (
+        component.get("contentType") == "disk-image"
+        or (component.get("contentGateway") or {}).get("contentType") == "disk-image"
+    )
+
     normalized_files = []
     if files_entries:
         logger.info(
             "  Processing %d files from files[] (Developer Portal):", len(files_entries)
         )
         for entry in files_entries:
-            normalized_source = _compress_file_entry(entry, "files", component_dir, ready_dir)
+            normalized_source = _compress_file_entry(
+                entry, "files", component_dir, ready_dir, is_disk_image_component=is_disk_image
+            )
             normalized_entry = dict(entry)
             # no-op for mac/linux, .zip correction for windows
             normalized_entry["source"] = normalized_source
@@ -191,7 +233,13 @@ def compress_component(component: dict, snapshot: dict) -> dict:
             "  Processing %d files from staged.files[] (Customer Portal):", len(staged_entries)
         )
         for entry in staged_entries:
-            _compress_file_entry(entry, "staged.files", component_dir, ready_dir)
+            _compress_file_entry(
+                entry,
+                "staged.files",
+                component_dir,
+                ready_dir,
+                is_disk_image_component=is_disk_image,
+            )
 
     updated_component = dict(component)
     if files_entries:

scripts/python/helpers/extract_artifacts.py

@@ -121,6 +121,46 @@ def _safe_extract_layer(
     return found
 
 
+def _extract_from_oras(
+    manifest: dict,
+    tmp_dir: Path,
+    wanted_files: list[str],
+    destination: Path,
+    component_name: str,
+) -> None:
+    """Copy raw ORAS blob layers to destination, matching by filename.
+
+    ORAS artifacts store raw file blobs as layers with an
+    ``org.opencontainers.image.title`` annotation containing the filename.
+    We match each wanted file (by basename) to its blob and copy it directly.
+    """
+    title_to_blob: dict[str, Path] = {}
+    for layer in manifest.get("layers", []):
+        title = (layer.get("annotations") or {}).get("org.opencontainers.image.title")
+        digest = layer.get("digest", "")
+        if title and digest:
+            blob_path = tmp_dir / digest.removeprefix("sha256:")
+            title_to_blob[title] = blob_path
+
+    logger.info(
+        "ORAS artifact detected for '%s'; available blobs: %s",
+        component_name,
+        list(title_to_blob),
+    )
+
+    for wanted in wanted_files:
+        basename = Path(wanted).name
+        blob = title_to_blob.get(basename)
+        if blob is None or not blob.is_file():
+            raise RuntimeError(
+                f"ORAS layer with title '{basename}' not found in component '{component_name}'. "
+                f"Available titles: {sorted(title_to_blob)}"
+            )
+        out = destination / basename
+        shutil.copy2(str(blob), str(out))
+        logger.info("Copied ORAS blob '%s' -> %s", basename, out)
+
+
 def process_component(component: dict) -> None:
     """Pull and extract one component's artifacts into CONTENT_DIR/<name>/."""
     name = component.get("name")
@@ -174,32 +214,42 @@ def process_component(component: dict) -> None:
         logger.info("Files to extract from RPA: %s", wanted_files)
 
         manifest = json.loads((tmp_dir / "manifest.json").read_text())
-        layer_digests = [layer["digest"] for layer in manifest.get("layers", [])]
-
-        for digest in layer_digests:
-            layer_file = tmp_dir / digest.removeprefix("sha256:")
-            if not layer_file.exists():
-                continue
-            with tarfile.open(str(layer_file)) as tf:
-                for image_path in extract_dirs:
-                    if _safe_extract_layer(tf, image_path, tmp_dir, layer_file.name):
-                        logger.info("Extracting %s/ from %s...", image_path, layer_file.name)
-                    else:
-                        logger.info(
-                            "skipping %s. It doesn't contain the %s dir",
-                            layer_file.name,
-                            image_path,
-                        )
-
-        for wanted in wanted_files:
-            src = tmp_dir / wanted
-            if src.is_file():
-                shutil.copy2(str(src), str(destination / src.name))
-            else:
-                logger.error("Expected file not found in container: %s", wanted)
-                raise RuntimeError(
-                    f"File '{wanted}' declared in RPA was not found in any container layer"
-                )
+
+        config_media_type = manifest.get("config", {}).get("mediaType", "")
+        if config_media_type == "application/vnd.oci.empty.v1+json":
+            # ORAS artifact: layers are raw file blobs, not tar archives.
+            # Each layer carries an org.opencontainers.image.title annotation
+            # that holds the original filename.  Copy blobs directly to destination.
+            _extract_from_oras(manifest, tmp_dir, wanted_files, destination, name)
+        else:
+            layer_digests = [layer["digest"] for layer in manifest.get("layers", [])]
+
+            for digest in layer_digests:
+                layer_file = tmp_dir / digest.removeprefix("sha256:")
+                if not layer_file.exists():
+                    continue
+                with tarfile.open(str(layer_file)) as tf:
+                    for image_path in extract_dirs:
+                        if _safe_extract_layer(tf, image_path, tmp_dir, layer_file.name):
+                            logger.info(
+                                "Extracting %s/ from %s...", image_path, layer_file.name
+                            )
+                        else:
+                            logger.info(
+                                "skipping %s. It doesn't contain the %s dir",
+                                layer_file.name,
+                                image_path,
+                            )
+
+            for wanted in wanted_files:
+                src = tmp_dir / wanted
+                if src.is_file():
+                    shutil.copy2(str(src), str(destination / src.name))
+                else:
+                    logger.error("Expected file not found in container: %s", wanted)
+                    raise RuntimeError(
+                        f"File '{wanted}' declared in RPA was not found in any container layer"
+                    )
     finally:
         shutil.rmtree(tmp_dir, ignore_errors=True)
 
@@ -227,6 +277,24 @@ def _matches(entry: dict, os_name: str) -> bool:
                     return True
             return False
 
+        is_disk_image_component = (
+            component.get("contentType") == "disk-image"
+            or (component.get("contentGateway") or {}).get("contentType") == "disk-image"
+        )
+
+        # Disk images must always be os: linux. Fail immediately if the RPA declares a
+        # disk-image component with macOS or Windows entries — those would route into
+        # the signing workflow and produce failures deep in the pipeline.
+        if is_disk_image_component:
+            for entry in all_file_entries:
+                entry_os = entry.get("os", "")
+                if entry_os in ("darwin", "windows"):
+                    raise RuntimeError(
+                        f"Component '{name}' has contentType: disk-image but entry "
+                        f"'{entry.get('source', '<unknown>')}' has os: {entry_os}. "
+                        f"Disk images must be os: linux. Fix the RPA before releasing."
+                    )
+
         if any(_matches(e, "darwin") for e in all_file_entries):
             (component_dir / "has_mac").touch()
             logger.info("  - macOS content detected")

scripts/python/helpers/push_artifacts.py

@@ -365,6 +365,19 @@ def run(exodus_gw_env: str, cgw_hostname: str, cert_expiration_warn_days: int) -
             cg = component.get("contentGateway") or {}
             cg["contentDir"] = str(component_dir)
             component["contentGateway"] = cg
+            # For disk-image components the RPA uses staged.files[] (not files[]) to
+            # describe the deliverables, but those files still need to appear in CGW.
+            # Merge staged.files into files[] so publish_to_cgw_wrapper registers them.
+            # NOTE: this intentionally mutates the component dict in-place. It is safe
+            # because the Pulp push and CDN exclusion logic for this component have already
+            # completed above, and the only remaining consumer is publish_to_cgw_wrapper
+            # called below via json.dumps(snapshot).
+            is_disk_image = (
+                component.get("contentType") == "disk-image"
+                or cg.get("contentType") == "disk-image"
+            )
+            if is_disk_image and not component.get("files"):
+                component["files"] = (component.get("staged") or {}).get("files", [])
 
     cgw_push = any(bool(c.get("contentGateway")) for c in snapshot.get("components", []))
     if cgw_push:

scripts/python/helpers/push_unsigned.py

@@ -40,6 +40,23 @@
 SUPPLEMENTARY_NAMES = {"readme", "license", "changelog"}
 SUPPLEMENTARY_EXTS = {".md", ".txt"}
 
+# Unambiguous disk-image file suffixes (simple and compound). Files matching
+# these are moved directly to the target directory without tar extraction even
+# when the component does not carry contentType: disk-image.
+# NOTE: .tar.gz is intentionally excluded — it is ambiguous between binary
+# archives and disk images (e.g. GCP images packaged as tarballs). Use
+# contentType: disk-image on the component to handle those cases.
+_DISK_IMAGE_SUFFIXES: frozenset[str] = frozenset(
+    {".qcow2", ".iso", ".iso.gz", ".raw.gz", ".vhd.gz"}
+)
+
+
+def _is_disk_image_file(filename: str) -> bool:
+    """Return True if *filename* has an unambiguous disk-image file suffix."""
+    lower = filename.lower()
+    return any(lower.endswith(ext) for ext in _DISK_IMAGE_SUFFIXES)
+
+
 logger = logging.getLogger(__name__)
 
 
@@ -80,8 +97,21 @@ def move_supplementary_out(src_root: Path, hold_root: Path) -> None:
             logger.info("  Held supplementary file: %s", rel)
 
 
-def _unpack_file_entries(entries: list[dict], component_dir: Path, unsigned_dir: Path) -> None:
-    """Extract each archive from entries into its OS/arch subdirectory under unsigned_dir."""
+def _unpack_file_entries(
+    entries: list[dict],
+    component_dir: Path,
+    unsigned_dir: Path,
+    *,
+    is_disk_image_component: bool = False,
+) -> None:
+    """Extract each archive from entries into its OS/arch subdirectory under unsigned_dir.
+
+    Files are moved directly (without unpacking) when either:
+    - *is_disk_image_component* is True (set when contentType: disk-image), or
+    - the filename has an unambiguous disk-image suffix (.qcow2, .iso, .iso.gz,
+      .raw.gz, .vhd.gz).
+    All other files are treated as tar archives and extracted.
+    """
     for entry in entries:
         source = entry.get("source", "")
         os_name = entry.get("os", "")
@@ -103,9 +133,12 @@ def _unpack_file_entries(entries: list[dict], component_dir: Path, unsigned_dir:
             continue
 
         target_dir.mkdir(parents=True, exist_ok=True)
-        with tarfile.open(str(archive_path)) as tf:
-            _safe_extract_archive(tf, target_dir, archive_name)
-        archive_path.unlink()
+        if is_disk_image_component or _is_disk_image_file(archive_name):
+            shutil.move(str(archive_path), str(target_dir / archive_name))
+        else:
+            with tarfile.open(str(archive_path)) as tf:
+                _safe_extract_archive(tf, target_dir, archive_name)
+            archive_path.unlink()
 
 
 def _safe_extract_archive(tf: tarfile.TarFile, target_dir: Path, archive_name: str) -> None:
@@ -157,9 +190,21 @@ def run(quay_url: str, pipeline_run_uid: str) -> None:
         if has_linux:
             (component_dir / "linux").mkdir(parents=True, exist_ok=True)
 
-        _unpack_file_entries(component.get("files") or [], component_dir, unsigned_dir)
+        is_disk_image = (
+            component.get("contentType") == "disk-image"
+            or (component.get("contentGateway") or {}).get("contentType") == "disk-image"
+        )
+        _unpack_file_entries(
+            component.get("files") or [],
+            component_dir,
+            unsigned_dir,
+            is_disk_image_component=is_disk_image,
+        )
         _unpack_file_entries(
-            (component.get("staged") or {}).get("files") or [], component_dir, unsigned_dir
+            (component.get("staged") or {}).get("files") or [],
+            component_dir,
+            unsigned_dir,
+            is_disk_image_component=is_disk_image,
         )
 
         supp_hold = component_dir / "supplementary"