Add support to interact with a local podman installation

pyproject.toml

@@ -10,12 +10,13 @@ packages = [{ include = "gardenlinux", from = "src" }]
 [tool.poetry.dependencies]
 python = "^3.13"
 apt-repo = "^0.5"
-boto3 = "^1.40.30"
+boto3 = "^1.40.43"
 click = "^8.2.1"
-cryptography = "^46.0.1"
+cryptography = "^46.0.2"
 jsonschema = "^4.25.1"
 networkx = "^3.5"
 oras = "^0.2.38"
+podman = "^5.6.0"
 pygit2 = "^1.18.2"
 pygments = "^2.19.2"
 PyYAML = "^6.0.2"
@@ -24,13 +25,13 @@ gitpython = "^3.1.45"
 [tool.poetry.group.dev.dependencies]
 bandit = "^1.8.6"
 black = "^25.1.0"
+isort = "^6.1.0"
 moto = "^5.1.12"
+pyright = "^1.1.406"
 python-dotenv = "^1.1.1"
 pytest = "^8.4.1"
 pytest-cov = "^7.0.0"
-isort = "^6.0.1"
 requests-mock = "^1.12.1"
-pyright = "^1.1.403"
 
 [tool.poetry.group.docs.dependencies]
 sphinx-rtd-theme = "^3.0.2"

src/gardenlinux/constants.py

@@ -153,11 +153,15 @@
 
 S3_DOWNLOADS_DIR = Path(os.path.dirname(__file__)) / ".." / "s3_downloads"
 
-GLVD_BASE_URL = "https://glvd.ingress.glvd.gardnlinux.shoot.canary.k8s-hana.ondemand.com/v1"
 GL_DEB_REPO_BASE_URL = "https://packages.gardenlinux.io/gardenlinux"
+GLVD_BASE_URL = (
+    "https://glvd.ingress.glvd.gardnlinux.shoot.canary.k8s-hana.ondemand.com/v1"
+)
 
 GARDENLINUX_GITHUB_RELEASE_BUCKET_NAME = "gardenlinux-github-releases"
 
+PODMAN_CONNECTION_MAX_IDLE_SECONDS = 3
+
 # https://github.com/gardenlinux/gardenlinux/issues/3044
 # Empty string is the 'legacy' variant with traditional root fs and still needed/supported
 IMAGE_VARIANTS = ["", "_usi", "_tpm2_trustedboot"]

src/gardenlinux/oci/__init__.py

@@ -9,5 +9,6 @@
 from .index import Index
 from .layer import Layer
 from .manifest import Manifest
+from .podman import Podman
 
-__all__ = ["Container", "ImageManifest", "Index", "Layer", "Manifest"]
+__all__ = ["Container", "ImageManifest", "Index", "Layer", "Manifest", "Podman"]

src/gardenlinux/oci/podman.py

@@ -0,0 +1,115 @@
+# -*- coding: utf-8 -*-
+
+"""
+OCI podman
+"""
+
+import logging
+from collections.abc import Mapping
+from os import PathLike
+from pathlib import Path
+from typing import Dict, Optional
+
+from .podman_context import PodmanContext
+from ..logger import LoggerSetup
+
+
+class Podman(object):
+    """
+    OCI podman provides access to an local podman installation.
+
+    :author:     Garden Linux Maintainers
+    :copyright:  Copyright 2024 SAP SE
+    :package:    gardenlinux
+    :subpackage: oci
+    :since:      0.11.0
+    :license:    https://www.apache.org/licenses/LICENSE-2.0
+                 Apache License, Version 2.0
+    """
+
+    def __init__(self, logger: Optional[logging.Logger] = None):
+        """
+        Constructor __init__(Podman)
+
+        :since: 0.11.0
+        """
+
+        if logger is None or not logger.hasHandlers():
+            logger = LoggerSetup.get_logger("gardenlinux.oci")
+
+        self._logger = logger
+
+    @PodmanContext.wrap
+    def build(
+        self,
+        build_path: str,
+        podman: PodmanContext,
+        build_args: Dict[str, str] = None,
+        oci_tag: str = None,
+        **kwargs,
+    ):
+        if isinstance(build_args, Mapping):
+            kwargs["buildargs"] = build_args
+
+        if oci_tag is not None:
+            kwargs["tag"] = oci_tag
+
+        image, _ = podman.images.build(path=build_path, dockerfile="Containerfile", **kwargs)
+        return image.id
+
+    @PodmanContext.wrap
+    def build_and_save_oci_archive(
+        self,
+        build_path: str,
+        oci_archive_file_name: str | PathLike[str],
+        podman: PodmanContext,
+        build_args: Dict[str, str] = None,
+        oci_tag: str = None,
+        **kwargs,
+    ):
+        oci_archive_file_name = Path(oci_archive_file_name)
+
+        image_id = self.build(build_path, podman=podman, build_args=build_args, oci_tag=oci_tag)
+        self.save_oci_archive(image_id, oci_archive_file_name, podman=podman)
+
+        return {oci_archive_file_name.name: image_id}
+
+    @PodmanContext.wrap
+    def load_oci_archives_from_directory(
+        self, oci_dir: str | PathLike[str], /, podman: PodmanContext
+    ) -> Dict[str, str]:
+        oci_archives = {}
+        oci_dir = Path(oci_dir)
+
+        for oci_archive in oci_dir.iterdir():
+            if not oci_archive.match("*.oci"):
+                continue
+
+            image = next(podman.images.load(file_path=oci_archive))
+            oci_archives[oci_archive.name] = image.id
+
+        return oci_archives
+
+    @PodmanContext.wrap
+    def save_oci_archive(
+        self,
+        image_id: str,
+        oci_archive_file_name: str | PathLike[str],
+        podman: PodmanContext,
+        oci_tag: str = None,
+    ):
+        oci_archive_file_name = Path(oci_archive_file_name)
+
+        if oci_archive_file_name.exists():
+            raise RuntimeError("OCI archive file does already exist")
+
+        image = podman.images.get(image_id)
+
+        with oci_archive_file_name.open("wb") as fp:
+            named = True
+
+            if oci_tag is not None:
+                named = oci_tag
+
+            for chunk in image.save(named=named):
+                fp.write(chunk)

src/gardenlinux/oci/podman_context.py

@@ -0,0 +1,172 @@
+# -*- coding: utf-8 -*-
+
+"""
+OCI podman context
+"""
+
+import logging
+from contextlib import ExitStack
+from functools import wraps
+from os import rmdir
+from pathlib import Path
+from subprocess import PIPE, Popen, STDOUT
+from tempfile import mkdtemp
+from time import sleep
+from typing import Any, Optional
+
+from podman.client import PodmanClient
+
+from ..constants import PODMAN_CONNECTION_MAX_IDLE_SECONDS
+from ..logger import LoggerSetup
+
+
+class PodmanContext(ExitStack):
+    """
+    OCI podman context provides a context manager to be used to interact with
+    the podman API from Python.
+
+    :author:     Garden Linux Maintainers
+    :copyright:  Copyright 2024 SAP SE
+    :package:    gardenlinux
+    :subpackage: oci
+    :since:      0.11.0
+    :license:    https://www.apache.org/licenses/LICENSE-2.0
+                 Apache License, Version 2.0
+    """
+
+    def __init__(self, logger: Optional[logging.Logger] = None):
+        """
+        Constructor __init__(PodmanContext)
+
+        :since: 0.11.0
+        """
+
+        ExitStack.__init__(self)
+
+        if logger is None or not logger.hasHandlers():
+            logger = LoggerSetup.get_logger("gardenlinux.oci")
+
+        self._logger = logger
+        self._podman = None
+        self._podman_daemon = None
+        self._tmpdir = None
+
+    def __enter__(self):
+        """
+        python.org: Enter the runtime context related to this object.
+
+        :return: (object) Podman context instance
+        :since: 0.11.0
+        """
+
+        self._tmpdir = mkdtemp()
+
+        podman_sock = str(Path(self._tmpdir, "podman.sock"))
+
+        self._podman = PodmanClient(base_url=f"unix://{podman_sock}")
+        self._podman_daemon = Popen(
+            args=[
+                "py-podman",
+                "system",
+                "service",
+                f"--time={PODMAN_CONNECTION_MAX_IDLE_SECONDS}",
+                f"unix://{podman_sock}",
+            ],
+            executable="podman",
+            stdout=PIPE,
+            stderr=STDOUT,
+        )
+
+        self.enter_context(self._podman_daemon)
+        self._wait_for_socket(podman_sock)
+        self.enter_context(self._podman)
+
+        return self
+
+    def __exit__(self, exc_type, exc_value, traceback):
+        """
+        python.org: Exit the runtime context related to this object.
+
+        :return: (bool) True to suppress exceptions
+        :since:  0.11.0
+        """
+
+        try:
+            self._podman_daemon.terminate()
+            self._podman_daemon.wait(PODMAN_CONNECTION_MAX_IDLE_SECONDS)
+
+            if exc_type is not None:
+                stdout = self._podman_daemon.stdout.read()
+                self._logger.error(
+                    f"Podman context encountered an error. Process output: {stdout}"
+                )
+        finally:
+            self._podman_daemon = None
+
+            rmdir(self._tmpdir)
+            self._tmpdir = None
+
+        return False
+
+    def __getattr__(self, name: str) -> Any:
+        """
+        python.org: Called when an attribute lookup has not found the attribute in
+        the usual places (i.e. it is not an instance attribute nor is it found in the
+        class tree for self).
+
+        :param name: Attribute name
+
+        :return: (mixed) Attribute
+        :since:  0.11.0
+        """
+
+        if self._podman_daemon is None:
+            raise RuntimeError("Podman context not ready")
+
+        return getattr(self._podman, name)
+
+    def _wait_for_socket(self, sock: str):
+        """
+        Waits for the socket file to be created.
+
+        :since: 0.11.0
+        """
+
+        sock_path = Path(sock)
+
+        for _ in range(0, 5 * PODMAN_CONNECTION_MAX_IDLE_SECONDS):
+            if sock_path.exists():
+                break
+
+            sleep(0.2)
+
+        if not sock_path.exists():
+            raise TimeoutError()
+
+    @staticmethod
+    def wrap(f) -> Any:
+        """
+        Wraps the given function to provide access to a podman client.
+
+        :since: 0.11.0
+        """
+
+        @wraps(f)
+        def decorator(*args, **kwargs) -> Any:
+            """
+            Decorator for wrapping a function or method with a call context.
+            """
+
+            if "podman" in kwargs:
+                if not isinstance(kwargs["podman"], PodmanContext):
+                    raise ValueError(
+                        "Podman context wrapped functions can not be called with `kwargs['podman']`"
+                    )
+
+                return f(*args, **kwargs)
+
+            with PodmanContext() as podman:
+                kwargs["podman"] = podman
+                return f(*args, **kwargs)
+
+        return decorator