Skip to content

feat(inspect): implement python-native skopeo inspect #34

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
merged 5 commits into from
May 26, 2025
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
27 changes: 27 additions & 0 deletions examples/image-inspect.py
Original file line number Diff line number Diff line change
@@ -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)
)
90 changes: 87 additions & 3 deletions image/client.py
Original file line number Diff line number Diff line change
Expand Up @@ -191,11 +191,11 @@ def query_blob(

Args:
str_or_ref (Union[str, ContainerImageReference]): An image reference corresponding to the blob descriptor
desc (Type[ContainerImageDescriptor]): A blob descriptor
desc (ContainerImageDescriptor): A blob descriptor
auth (Dict[str, Any]): A valid docker config JSON loaded into a dict

Returns:
Type[requests.Response]: The registry API blob response
requests.Response: The registry API blob response
"""
# If given a str, then load as a ref
ref = str_or_ref
Expand Down Expand Up @@ -265,7 +265,91 @@ def get_config(
# Load the manifest into a dict and return
config = res.json()
return config


@staticmethod
def query_tags(
str_or_ref: Union[str, ContainerImageReference],
auth: Dict[str, Any]
) -> requests.Response:
"""
Fetches the list of tags for a reference from the registry API and
returns as a dict
Args:
str_or_ref (Union[str, ContainerImageReference]): An image reference
auth (Dict[str, Any]): A valid docker config JSON loaded into a dict
Returns:
requests.Response: The registry API tag list response
"""
# If given a str, then load as a ref
ref = str_or_ref
if isinstance(str_or_ref, str):
ref = ContainerImageReference(str_or_ref)

# Construct the API URL for querying the image manifest
api_base_url = ContainerImageRegistryClient.get_registry_base_url(
ref
)
image_identifier = ref.get_identifier()
api_url = f'{api_base_url}/tags/list'

# Construct the headers for querying the image manifest
headers = {
'Accept': 'application/json'
}

# Get the matching auth for the image from the docker config JSON
reg_auth, found = ContainerImageRegistryClient.get_registry_auth(
ref,
auth
)
if found:
headers['Authorization'] = f'Basic {reg_auth}'

# Send the request to the distribution registry API
# If it fails with a 401 response code and auth given, do OAuth dance
res = requests.get(api_url, headers=headers)
if res.status_code == 401 and \
'www-authenticate' in res.headers.keys():
# Do Oauth dance if basic auth fails
# Ref: https://distribution.github.io/distribution/spec/auth/token/
scheme, token = ContainerImageRegistryClient.get_auth_token(
res, reg_auth
)
headers['Authorization'] = f'{scheme} {token}'
res = requests.get(api_url, headers=headers)

# Raise exceptions on error status codes
res.raise_for_status()
return res

@staticmethod
def list_tags(
str_or_ref: Union[str, ContainerImageReference],
auth: Dict[str, Any]
) -> Dict[str, Any]:
"""
Fetches the list of tags for a reference from the registry API and
returns as a dict
Args:
str_or_ref (Union[str, ContainerImageReference]): An image reference
auth (Dict[str, Any]): A valid docker config JSON loaded into a dict
Returns:
Dict[str, Any]: The config as a dict
"""
# If given a str, then load as a ref
ref = str_or_ref
if isinstance(str_or_ref, str):
ref = ContainerImageReference(str_or_ref)

# Query the tags, get the tag list response
res = ContainerImageRegistryClient.query_tags(
ref, auth
)

# Load the tag list into a dict and return
tags = res.json()
return tags

@staticmethod
def query_manifest(
str_or_ref: Union[str, ContainerImageReference],
Expand Down
50 changes: 48 additions & 2 deletions image/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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.get_runtime_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", "")
Loading
Loading