Extend usage of Podman to analyze OCI image layers

.github/actions/setup/action.yml

@@ -4,7 +4,7 @@ description: Installs the given GardenLinux Python library
 inputs:
   version:
     description: GardenLinux Python library version
-    default: "0.10.16"
+    default: "0.10.19"
   python_version:
     description: Python version to setup
     default: "3.13"

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "gardenlinux"
-version = "0.10.16"
+version = "0.10.19"
 description = "Contains tools to work with the features directory of gardenlinux, for example deducting dependencies from feature sets or validating cnames"
 authors = ["Garden Linux Maintainers <contact@gardenlinux.io>"]
 license = "Apache-2.0"

src/gardenlinux/constants.py

@@ -159,14 +159,18 @@
 
 RELEASE_ID_FILE = ".github_release_id"
 
-REQUESTS_TIMEOUTS = (5, 30)  # connect, read
+REQUESTS_TIMEOUTS = (5, 60)  # connect, read
 
 S3_DOWNLOADS_DIR = Path(os.path.dirname(__file__)) / ".." / "s3_downloads"
 
 GARDENLINUX_GITHUB_RELEASE_BUCKET_NAME = "gardenlinux-github-releases"
 GLVD_BASE_URL = "https://security.gardenlinux.org/v1"
 
 PODMAN_CONNECTION_MAX_IDLE_SECONDS = 3
+PODMAN_FS_CHANGE_ADDED = "added"
+PODMAN_FS_CHANGE_DELETED = "deleted"
+PODMAN_FS_CHANGE_MODIFIED = "modified"
+PODMAN_FS_CHANGE_UNSUPPORTED = "unsupported"
 
 # https://github.com/gardenlinux/gardenlinux/issues/3044
 # Empty string is the 'legacy' variant with traditional root fs and still needed/supported

src/gardenlinux/features/reproducibility/comparator.py

@@ -9,11 +9,15 @@
 import importlib.resources
 import json
 import re
-import tarfile
-import tempfile
 from os import PathLike
 from pathlib import Path
-from typing import Optional
+
+from ...constants import (
+    PODMAN_FS_CHANGE_ADDED,
+    PODMAN_FS_CHANGE_DELETED,
+    PODMAN_FS_CHANGE_MODIFIED,
+)
+from ...oci import Image, Podman, PodmanContext
 
 
 class Comparator(object):
@@ -29,120 +33,27 @@ class Comparator(object):
                  Apache License, Version 2.0
     """
 
-    _default_whitelist: list[str] = []
-
-    _nightly_whitelist = json.loads(
-        importlib.resources.read_text(__name__, "nightly_whitelist.json")
-    )
-
-    def __init__(
-        self, nightly: bool = False, whitelist: list[str] = _default_whitelist
-    ):
+    def __init__(self, nightly: bool = False, whitelist: list[str] = []):
         """
         Constructor __init__(Comparator)
 
         :param nightly:                 Flag indicating if the nightlywhitelist should be used
         :param whitelst:                Additional whitelist
 
-        :since: 1.0.0
-        """
-        self.whitelist = whitelist
-        if nightly:
-            self.whitelist += self._nightly_whitelist
-
-    @staticmethod
-    def _unpack(file: PathLike[str]) -> tempfile.TemporaryDirectory[str]:
-        """
-        Unpack a .tar archive or .oci image into a temporary dictionary
-
-        :param file:                    .tar or .oci file
-
-        :return: TemporaryDirectory     Temporary directory containing the unpacked file
         :since: 1.0.0
         """
 
-        output_dir = tempfile.TemporaryDirectory()
-        file = Path(file).resolve()
-        if file.name.endswith(".oci"):
-            with tempfile.TemporaryDirectory() as extracted:
-                # Extract .oci file
-                with tarfile.open(file, "r") as tar:
-                    tar.extractall(
-                        path=extracted, filter="fully_trusted", members=tar.getmembers()
-                    )
-
-                layers_dir = Path(extracted).joinpath("blobs/sha256")
-                assert layers_dir.is_dir()
-
-                with open(Path(extracted).joinpath("index.json"), "r") as f:
-                    index = json.load(f)
-
-                # Only support first manifest
-                manifest = index["manifests"][0]["digest"].split(":")[1]
-
-                with open(layers_dir.joinpath(manifest), "r") as f:
-                    manifest = json.load(f)
-
-                layers = [layer["digest"].split(":")[1] for layer in manifest["layers"]]
-
-                # Extract layers in order
-                for layer in layers:
-                    layer_path = layers_dir.joinpath(layer)
-                    if tarfile.is_tarfile(layer_path):
-                        with tarfile.open(layer_path, "r") as tar:
-                            for member in tar.getmembers():
-                                try:
-                                    tar.extract(
-                                        member,
-                                        path=output_dir.name,
-                                        filter="fully_trusted",
-                                    )
-                                except tarfile.AbsoluteLinkError:
-                                    # Convert absolute link to relative link
-                                    member.linkpath = (
-                                        "../" * member.path.count("/")
-                                        + member.linkpath[1:]
-                                    )
-                                    tar.extract(
-                                        member,
-                                        path=output_dir.name,
-                                        filter="fully_trusted",
-                                    )
-                                except tarfile.TarError as e:
-                                    print(f"Skipping {member.name} due to error: {e}")
-        else:
-            with tarfile.open(file, "r") as tar:
-                tar.extractall(
-                    path=output_dir.name,
-                    filter="fully_trusted",
-                    members=tar.getmembers(),
-                )
-
-        return output_dir
-
-    def _diff_files(
-        self, cmp: filecmp.dircmp[str], left_root: Optional[Path] = None
-    ) -> list[str]:
-        """
-        Recursively compare files
-
-        :param cmp:                     Dircmp to recursively compare
-        :param left_root:               Left root to obtain the archive relative path
-
-        :return: list[Path]             List of paths with different content
-        :since: 1.0.0
-        """
-
-        result = []
-        if not left_root:
-            left_root = Path(cmp.left)
-        for name in cmp.diff_files:
-            result.append(f"/{Path(cmp.left).relative_to(left_root).joinpath(name)}")
-        for sub_cmp in cmp.subdirs.values():
-            result += self._diff_files(sub_cmp, left_root=left_root)
-        return result
+        self.whitelist = whitelist
 
-    def generate(self, a: PathLike[str], b: PathLike[str]) -> tuple[list[str], bool]:
+        if nightly:
+            self.whitelist += json.loads(
+                importlib.resources.read_text(__name__, "nightly_whitelist.json")
+            )
+
+    @PodmanContext.wrap
+    def generate(
+        self, a: PathLike[str], b: PathLike[str], podman: PodmanContext
+    ) -> tuple[list[str], bool]:
         """
         Compare two .tar/.oci images with each other
 
@@ -156,16 +67,51 @@ def generate(self, a: PathLike[str], b: PathLike[str]) -> tuple[list[str], bool]
         if filecmp.cmp(a, b, shallow=False):
             return [], False
 
-        with self._unpack(a) as unpacked_a, self._unpack(b) as unpacked_b:
-            cmp = filecmp.dircmp(unpacked_a, unpacked_b, shallow=False)
-
-            diff_files = self._diff_files(cmp)
-
-        filtered = [
-            file
-            for file in diff_files
-            if not any(re.match(pattern, file) for pattern in self.whitelist)
-        ]
-        whitelist = len(diff_files) != len(filtered)
-
-        return filtered, whitelist
+        a = Path(a)
+        a_image_id = None
+
+        b = Path(b)
+        b_image_id = None
+
+        differences = []
+        podman_api = Podman()
+
+        try:
+            if a.suffix == ".oci":
+                a_image_id = podman_api.load_oci_archive(a, podman=podman)
+            elif a.suffix == ".tar":
+                a_image_id = Image.import_plain_tar(a, podman=podman)
+            else:
+                raise RuntimeError(f"Unsupported file type for comparison: {a.name}")
+
+            if b.suffix == ".oci":
+                b_image_id = podman_api.load_oci_archive(b, podman=podman)
+            elif b.suffix == ".tar":
+                b_image_id = Image.import_plain_tar(b, podman=podman)
+            else:
+                raise RuntimeError(f"Unsupported file type for comparison: {b.name}")
+
+            image = podman_api.get_image(a_image_id, podman=podman)
+
+            result = image.get_filesystem_changes(
+                parent_layer_image_id=b_image_id, podman=podman
+            )
+
+            differences = (
+                result[PODMAN_FS_CHANGE_ADDED] + result[PODMAN_FS_CHANGE_DELETED]
+            )
+
+            whitelist = False
+
+            for entry in result[PODMAN_FS_CHANGE_MODIFIED]:
+                if not any(re.match(pattern, entry) for pattern in self.whitelist):
+                    differences.append(entry)
+                else:
+                    whitelist = True
+        finally:
+            if a_image_id is not None:
+                podman.images.remove(a_image_id)
+            if b_image_id is not None:
+                podman.images.remove(b_image_id)
+
+        return differences, whitelist

src/gardenlinux/github/release/__init__.py

@@ -1,11 +1,7 @@
-import json
 import logging
-import os
 import sys
 
-import requests
-
-from ...constants import RELEASE_ID_FILE, REQUESTS_TIMEOUTS
+from ...constants import RELEASE_ID_FILE
 from ...logger import LoggerSetup
 from .release import Release
 
@@ -63,47 +59,4 @@ def write_to_release_id_file(release_id: str | int) -> None:
         sys.exit(1)
 
 
-def upload_to_github_release_page(
-    github_owner: str,
-    github_repo: str,
-    gardenlinux_release_id: str | int,
-    file_to_upload: str,
-    dry_run: bool,
-) -> None:
-    if dry_run:
-        LOGGER.info(
-            f"Dry run: would upload {file_to_upload} to release {gardenlinux_release_id} in repo {github_owner}/{github_repo}"
-        )
-        return
-
-    token = os.environ.get("GITHUB_TOKEN")
-    if not token:
-        raise ValueError("GITHUB_TOKEN environment variable not set")
-
-    headers = {
-        "Authorization": f"token {token}",
-        "Content-Type": "application/octet-stream",
-    }
-
-    upload_url = f"https://uploads.github.com/repos/{github_owner}/{github_repo}/releases/{gardenlinux_release_id}/assets?name={os.path.basename(file_to_upload)}"
-
-    try:
-        with open(file_to_upload, "rb") as f:
-            file_contents = f.read()
-    except IOError as e:
-        LOGGER.error(f"Error reading file {file_to_upload}: {e}")
-        return
-
-    response = requests.post(
-        upload_url, headers=headers, data=file_contents, timeout=REQUESTS_TIMEOUTS
-    )
-    if response.status_code == 201:
-        LOGGER.info("Upload successful")
-    else:
-        LOGGER.error(
-            f"Upload failed with status code {response.status_code}: {response.text}"
-        )
-        response.raise_for_status()
-
-
-__all__ = ["Release", "write_to_release_id_file", "upload_to_github_release_page"]
+__all__ = ["Release", "write_to_release_id_file"]

src/gardenlinux/github/release/__main__.py

@@ -6,7 +6,6 @@
 
 from ..release_notes import create_github_release_notes
 from . import (
-    upload_to_github_release_page,
     write_to_release_id_file,
 )
 from .release import Release
@@ -149,6 +148,10 @@ def get_parser() -> argparse.ArgumentParser:
         help="Perform a dry run without actually uploading the file.",
     )
 
+    upload_parser.add_argument(
+        "--overwrite-same-name", action="store_true", default=False
+    )
+
     return parser
 
 
@@ -169,6 +172,7 @@ def main() -> None:
         body = create_github_release_notes(
             args.tag, args.commit, GARDENLINUX_GITHUB_RELEASE_BUCKET_NAME
         )
+
         if args.dry_run:
             print("Dry Run ...")
             print("This release would be created:")
@@ -184,9 +188,16 @@ def main() -> None:
             write_to_release_id_file(f"{release_id}")
             LOGGER.info(f"Release created with ID: {release_id}")
     elif args.command == "upload":
-        upload_to_github_release_page(
-            args.owner, args.repo, args.release_id, args.file_path, args.dry_run
-        )
+        release = Release.get(args.release_id, repo=args.repo, owner=args.owner)
+
+        if args.dry_run:
+            print("Dry Run ...")
+
+            print(
+                f"The file {args.file_path} would be uploaded for release: {release.name}"
+            )
+        else:
+            release.upload_asset(args.file_path, args.overwrite_same_name)
     else:
         parser.print_help()