feat(inspect): implement python-native skopeo inspect

examples/image-inspect.py

@@ -0,0 +1,27 @@
+######
+# Hack
+#
+# Make sibling modules visible to this nested executable
+import os, sys
+sys.path.insert(
+    0,
+    os.path.dirname(
+        os.path.dirname(
+            os.path.realpath(__file__)
+        )
+    )
+)
+# End Hack
+######
+
+from image.containerimage import ContainerImage
+
+# Initialize a ContainerImage given a tag reference
+my_image = ContainerImage("registry.k8s.io/pause:3.5")
+
+# Display the inspect information for the container image
+my_image_inspect = my_image.inspect(auth={})
+print(
+    f"Inspect of {str(my_image)}: \n" + \
+    str(my_image_inspect)
+)

image/config.py

@@ -3,7 +3,7 @@
 configuration for a container image.
 """
 
-from typing import Dict, Any, Tuple, Type, Union
+from typing import Dict, Any, Tuple, Type, Union, List
 from jsonschema import validate, ValidationError
 from image.configschema import CONTAINER_IMAGE_CONFIG_SCHEMA
 from image.platform import ContainerImagePlatform
@@ -105,4 +105,50 @@ def get_platform(self) -> Type[ContainerImagePlatform]:
         variant = self.get_variant()
         if variant != None:
             platform_dict["variant"] = variant
-        return ContainerImagePlatform(platform_dict)
+        return ContainerImagePlatform(platform_dict)
+
+    def get_labels(self) -> Dict[str, str]:
+        """
+        Returns the container image labels from the config
+
+        Returns:
+            Dict[str, str]: The labels from the config
+        """
+        return self.config.get("Labels", {})
+
+    def get_created_date(self) -> str:
+        """
+        Returns the created date of the container image from the config
+
+        Returns:
+            str: The created date, as a string
+        """
+        return self.config.get("created", "")
+
+    def get_runtime_config(self) -> Dict[str, Any]:
+        """
+        Returns the runtime config for the container image from its config
+
+        Returns:
+            Dict[str, Any]: The container image runtime config
+        """
+        return self.config.get("config", {})
+
+    def get_env(self) -> List[str]:
+        """
+        Returns the list of environment variables set for the container image
+        at build time from the container image runtime config
+
+        Returns:
+            List[str]: The list of environment variables
+        """
+        return self.get_runtime_config().get("Env", [])
+
+    def get_author(self) -> str:
+        """
+        Returns the author of the container image from its config
+
+        Returns:
+            str: The container image author
+        """
+        return self.config.get("Author", "")

image/containerimage.py

@@ -9,20 +9,21 @@
 from __future__ import annotations
 import json
 import requests
-from typing                 import  List, Dict, Any, \
-                                    Union, Type, Iterator
-from image.byteunit         import  ByteUnit
-from image.client           import  ContainerImageRegistryClient
-from image.config           import  ContainerImageConfig
-from image.errors           import  ContainerImageError
-from image.manifestfactory  import  ContainerImageManifestFactory
-from image.manifestlist     import  ContainerImageManifestList
-from image.oci              import  ContainerImageManifestOCI, \
-                                    ContainerImageIndexOCI
-from image.platform         import  ContainerImagePlatform
-from image.reference        import  ContainerImageReference
-from image.v2s2             import  ContainerImageManifestV2S2, \
-                                    ContainerImageManifestListV2S2
+from typing                         import  List, Dict, Any, \
+                                            Union, Type, Iterator
+from image.byteunit                 import  ByteUnit
+from image.client                   import  ContainerImageRegistryClient
+from image.config                   import  ContainerImageConfig
+from image.containerimageinspect    import  ContainerImageInspect
+from image.errors                   import  ContainerImageError
+from image.manifestfactory          import  ContainerImageManifestFactory
+from image.manifestlist             import  ContainerImageManifestList
+from image.oci                      import  ContainerImageManifestOCI, \
+                                            ContainerImageIndexOCI
+from image.platform                 import  ContainerImagePlatform
+from image.reference                import  ContainerImageReference
+from image.v2s2                     import  ContainerImageManifestV2S2, \
+                                            ContainerImageManifestListV2S2
 
 #########################################
 # Classes for managing container images #
@@ -85,6 +86,103 @@ def is_oci_static(
         return isinstance(manifest, ContainerImageManifestOCI) or \
                 isinstance(manifest, ContainerImageIndexOCI)
 
+    @staticmethod
+    def get_host_platform_manifest_static(
+            ref: ContainerImageReference,
+            manifest: Union[
+                ContainerImageManifestV2S2,
+                ContainerImageManifestListV2S2,
+                ContainerImageManifestOCI,
+                ContainerImageIndexOCI
+            ],
+            auth: Dict[str, Any]
+        ) -> Union[
+            ContainerImageManifestV2S2,
+            ContainerImageManifestOCI
+        ]:
+        """
+        Given an image's reference and manifest, this static method checks if
+        the manifest is a manifest list, and attempts to get the manifest from
+        the list matching the host platform.
+
+        Args:
+            ref (ContainerImageReference): The image reference corresponding to the manifest
+            manifest (Union[ContainerImageManifestV2S2,ContainerImageManifestListV2S2,ContainerImageManifestOCI,ContainerImageIndexOCI]): The manifest object, generally from get_manifest method
+            auth (Dict[str, Any]): A valid docker config JSON with auth into the ref's registry
+
+        Returns:
+            Union[ContainerImageManifestV2S2,ContainerImageManifestOCI]: The manifest response from the registry API
+
+        Raises:
+            ContainerImageError: Error if the image is a manifest list without a manifest matching the host platform
+        """
+        host_manifest = manifest
+
+        # If manifest list, get the manifest matching the host platform
+        if ContainerImage.is_manifest_list_static(manifest):
+            found = False
+            host_entry_digest = None
+            host_plt = ContainerImagePlatform.get_host_platform()
+            entries = manifest.get_entries()
+            for entry in entries:
+                if entry.get_platform() == host_plt:
+                    found = True
+                    host_entry_digest = entry.get_digest()
+            if not found:
+                raise ContainerImageError(
+                    "no image found in manifest list for platform: " + \
+                    f"{str(host_plt)}"
+                )
+            host_ref = ContainerImage(
+                f"{ref.get_name()}@{host_entry_digest}"
+            )
+            host_manifest = host_ref.get_manifest(auth=auth)
+
+        # Return the manifest matching the host platform
+        return host_manifest
+
+    @staticmethod
+    def get_config_static(
+            ref: ContainerImageReference,
+            manifest: Union[
+                ContainerImageManifestV2S2,
+                ContainerImageManifestListV2S2,
+                ContainerImageManifestOCI,
+                ContainerImageIndexOCI
+            ],
+            auth: Dict[str, Any]
+        ) -> ContainerImageConfig:
+        """
+        Given an image's manifest, this static method fetches that image's
+        config from the distribution registry API.  If the image is a manifest
+        list, then it gets the config corresponding to the manifest matching
+        the host platform.
+
+        Args:
+            ref (ContainerImageReference): The image reference corresponding to the manifest
+            manifest (Union[ContainerImageManifestV2S2,ContainerImageManifestListV2S2,ContainerImageManifestOCI,ContainerImageIndexOCI]): The manifest object, generally from get_manifest method
+            auth (Dict[str, Any]): A valid docker config JSON with auth into this image's registry
+
+        Returns:
+            ContainerImageConfig: The config for this image
+
+        Raises:
+            ContainerImageError: Error if the image is a manifest list without a manifest matching the host platform
+        """
+        # If manifest list, get the manifest matching the host platform
+        manifest = ContainerImage.get_host_platform_manifest_static(
+            ref, manifest, auth
+        )
+
+        # Get the image's config
+        return ContainerImageConfig(
+            ContainerImageRegistryClient.get_config(
+                ref,
+                manifest.get_config_descriptor(),
+                auth=auth
+            )
+        )
+
     def __init__(self, ref: str):
         """
         Constructor for the ContainerImage class
@@ -179,6 +277,61 @@ def get_manifest(self, auth: Dict[str, Any]) -> Union[
             ContainerImageRegistryClient.get_manifest(self, auth)
         )
 
+    def get_host_platform_manifest(self, auth: Dict[str, Any]) -> Union[
+            ContainerImageManifestOCI,
+            ContainerImageManifestV2S2
+        ]:
+        """
+        Fetches the manifest from the distribution registry API.  If the
+        manifest is a manifest list, then it attempts to fetch the manifest
+        in the list matching the host platform.  If not found, an exception is
+        raised.
+
+        Args:
+            auth (Dict[str, Any]): A valid docker config JSON with auth into this image's registry
+
+        Returns:
+            Union[ContainerImageManifestV2S2,ContainerImageManifestOCI]: The manifest response from the registry API
+
+        Raises:
+            ContainerImageError: Error if the image is a manifest list without a manifest matching the host platform
+        """
+        # Get the container image's manifest
+        manifest = self.get_manifest(auth=auth)
+
+        # Return the host platform manifest
+        return ContainerImage.get_host_platform_manifest_static(
+            self,
+            manifest,
+            auth
+        )
+
+    def get_config(self, auth: Dict[str, Any]) -> ContainerImageConfig:
+        """
+        Fetches the image's config from the distribution registry API.  If the
+        image is a manifest list, then it gets the config corresponding to the
+        manifest matching the host platform.
+
+        Args:
+            auth (Dict[str, Any]): A valid docker config JSON with auth into this image's registry
+
+        Returns:
+            ContainerImageConfig: The config for this image
+
+        Raises:
+            ContainerImageError: Error if the image is a manifest list without a manifest matching the host platform
+        """
+        # Get the image's manifest
+        manifest = self.get_manifest(auth=auth)
+
+        # Use the image's manifest to get the image's config
+        config = ContainerImage.get_config_static(
+            self, manifest, auth
+        )
+
+        # Return the image's config
+        return config
+
     def exists(self, auth: Dict[str, Any]) -> bool:
         """
         Determine if the image reference corresponds to an image in the remote
@@ -266,6 +419,60 @@ def get_size_formatted(self, auth: Dict[str, Any]) -> str:
         """
         return ByteUnit.format_size_bytes(self.get_size(auth))
 
+    def inspect(self, auth: Dict[str, Any]) -> ContainerImageInspect:
+        """
+        Returns a collection of basic information about the image, equivalent
+        to skopeo inspect.
+
+        Args:
+            auth (Dict[str, Any]): A valid docker config JSON loaded into a dict
+
+        Returns:
+            ContainerImageInspect: A collection of information about the image
+        """
+        # Get the image's manifest
+        manifest = self.get_host_platform_manifest(auth=auth)
+
+        # Use the image's manifest to get the image's config
+        config = ContainerImage.get_config_static(
+            self, manifest, auth
+        )
+
+        # Format the inspect dictionary
+        inspect = {
+            "Name": self.get_name(),
+            "Digest": self.get_digest(auth=auth),
+            # TODO: Implement v2s1 manifest extension - only v2s1 manifests use this value
+            "DockerVersion": "",
+            "Created": config.get_created_date(),
+            "Labels": config.get_labels(),
+            "Architecture": config.get_architecture(),
+            "Variant": config.get_variant() or "",
+            "Os": config.get_os(),
+            "Layers": [ 
+                layer.get_digest() \
+                for layer \
+                in manifest.get_layer_descriptors()
+            ],
+            "LayersData": [
+                {
+                    "MIMEType": layer.get_media_type(),
+                    "Digest": layer.get_digest(),
+                    "Size": layer.get_size(),
+                    "Annotations": layer.get_annotations() or {}
+                } for layer in manifest.get_layer_descriptors()
+            ],
+            "Env": config.get_env(),
+            "Author": config.get_author()
+        }
+
+        # Set the tag in the inspect dict
+        if self.is_tag_ref():
+            inspect["Tag"] = self.get_identifier()
+
+        # TODO: Get the RepoTags for the image
+        return ContainerImageInspect(inspect)
+
     def delete(self, auth: Dict[str, Any]):
         """
         Deletes the image from the registry.