Add initial support to interact with a local podman installation

Signed-off-by: Tobias Wolf <[email protected]>

Author: NotTheEvilOne

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"
+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"
 
 GARDENLINUX_GITHUB_RELEASE_BUCKET_NAME = "gardenlinux-github-releases"
 
+PODMAN_CONNECTION_MAX_DURATION = 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,55 @@
+# -*- coding: utf-8 -*-
+
+"""
+OCI podman
+"""
+
+import logging
+from os import PathLike
+from pathlib import Path
+from typing import 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 load_oci_archives_from_directory(
+        self, oci_dir: str | PathLike[str], /, podman: PodmanContext
+    ):
+        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

src/gardenlinux/oci/podman_context.py

@@ -0,0 +1,169 @@
+# -*- 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 Optional
+
+from podman.client import PodmanClient
+
+from ..constants import PODMAN_CONNECTION_MAX_DURATION
+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_DURATION}",
+                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_DURATION)
+
+            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):
+        """
+        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_DURATION):
+            if sock_path.exists():
+                break
+
+            sleep(0.2)
+
+        if not sock_path.exists():
+            raise TimeoutError()
+
+    @staticmethod
+    def wrap(f):
+        """
+        Wraps the given function to provide access to a podman client.
+
+        :since: 0.11.0
+        """
+
+        @wraps(f)
+        def decorator(*args, **kwargs):
+            """
+            Decorator for wrapping a function or method with a call context.
+            """
+
+            if "podman" in kwargs:
+                raise ValueError(
+                    "Podman context wrapped functions can not be called with `kwargs['podman']`"
+                )
+
+            with PodmanContext() as podman:
+                kwargs["podman"] = podman
+                return f(*args, **kwargs)
+
+        return decorator