spack.oci package

Submodules

spack.oci.image module

class spack.oci.image.Digest(*, algorithm: str, digest: str)

Bases: object

Represents a digest in the format <algorithm>:<digest>. Currently only supports sha256 digests.

algorithm

digest

classmethod from_sha256(digest: str) → Digest

classmethod from_string(string: str) → Digest

class spack.oci.image.ImageReference(*, domain: str, name: str, tag: str = 'latest', digest: Digest | None = None, scheme: str = 'https')

Bases: object

A parsed image of the form domain/name:tag[@digest]. The digest is optional, and domain and tag are automatically filled out with defaults when parsed from string.

blob_url(digest: str | Digest) → str

digest

domain

endpoint(path: str = '') → str

classmethod from_string(string: str, *, scheme: str = 'https') → ImageReference

classmethod from_url(url: str) → ImageReference

Parse an OCI URL into an ImageReference, either oci:// or oci+http://.

manifest_url() → str

name

scheme

tag

tags_url() → str

uploads_url(digest: Digest | None = None) → str

with_digest(digest: str | Digest) → ImageReference

with_tag(tag: str) → ImageReference

spack.oci.image.default_config(architecture: str, os: str)

spack.oci.image.default_manifest()

spack.oci.image.ensure_sha256_checksum(oci_blob: str)

Validate that the reference is of the format sha256:<checksum> Return the checksum if valid, raise ValueError otherwise.

spack.oci.image.ensure_valid_tag(tag: str) → str

Ensure a tag is valid for an OCI registry.

spack.oci.image.is_oci_url(url: str) → bool

Check if the URL is an OCI URL.

spack.oci.oci module

class spack.oci.oci.Blob(compressed_digest, uncompressed_digest, size)

Bases: NamedTuple

compressed_digest: Digest

Alias for field number 0

size: int

Alias for field number 2

uncompressed_digest: Digest

Alias for field number 1

spack.oci.oci.all_content_type

All OCI manifest / index content types

spack.oci.oci.blob_exists(ref: ImageReference, digest: Digest, _urlopen: Callable[[...], HTTPResponse] | None = None) → bool

Checks if a blob exists in an OCI registry

spack.oci.oci.copy_missing_layers(src: ImageReference, dst: ImageReference, architecture: str, _urlopen: Callable[[...], HTTPResponse] | None = None) → Tuple[dict, dict]

Copy image layers from src to dst for given architecture.

Parameters:

• src – The source image reference.

• dst – The destination image reference.

• architecture – The architecture (when referencing an index)

Returns:

Tuple of manifest and config of the base image.

spack.oci.oci.copy_missing_layers_with_retry(src: ImageReference, dst: ImageReference, architecture: str, _urlopen: Callable[[...], HTTPResponse] | None = None) → Tuple[dict, dict]

Same as copy_missing_layers, but with retry wrapper

spack.oci.oci.get_manifest_and_config(ref: ImageReference, architecture='amd64', recurse=3, _urlopen: Callable[[...], HTTPResponse] | None = None) → Tuple[dict, dict]

Recursively fetch manifest and config for a given image reference with a given architecture.

Parameters:

• ref – The image reference.

• architecture – The architecture (when referencing an index)

• recurse – How many levels of index to recurse into.

Returns:

A tuple of (manifest, config)

spack.oci.oci.get_manifest_and_config_with_retry(ref: ImageReference, architecture='amd64', recurse=3, _urlopen: Callable[[...], HTTPResponse] | None = None) → Tuple[dict, dict]

Same as get_manifest_and_config, but with retry wrapper

spack.oci.oci.image_from_mirror(mirror: Mirror) → ImageReference

Given an OCI based mirror, extract the URL and image name from it

spack.oci.oci.index_content_type

OCI index content types (including docker type)

spack.oci.oci.list_tags(ref: ImageReference, _urlopen: Callable[[...], HTTPResponse] | None = None) → List[str]

Retrieves the list of tags associated with an image, handling pagination.

spack.oci.oci.make_stage(url: str, digest: Digest, keep: bool = False, _urlopen: Callable[[...], HTTPResponse] | None = None) → Stage

spack.oci.oci.manifest_content_type

OCI manifest content types (including docker type)

spack.oci.oci.upload_blob(ref: ImageReference, file: str, digest: Digest, force: bool = False, small_file_size: int = 0, _urlopen: Callable[[...], HTTPResponse] | None = None) → bool

Uploads a blob to an OCI registry

We only do monolithic uploads, even though it’s very simple to do chunked. Observed problems with chunked uploads: (1) it’s slow, many sequential requests, (2) some registries set an unknown max chunk size, and the spec doesn’t say how to obtain it

Parameters:

• ref – The image reference.

• file – The file to upload.

• digest – The digest of the file.

• force – Whether to force upload the blob, even if it already exists.

• small_file_size – For files at most this size, attempt to do a single POST request instead of POST + PUT. Some registries do no support single requests, and others do not specify what size they support in single POST. For now this feature is disabled by default (0KB)

Returns:

True if the blob was uploaded, False if it already existed.

spack.oci.oci.upload_blob_with_retry(ref: ImageReference, file: str, digest: Digest, force: bool = False, small_file_size: int = 0, _urlopen: Callable[[...], HTTPResponse] | None = None) → bool

Same as upload_blob, but with retry wrapper

spack.oci.oci.upload_manifest(ref: ImageReference, manifest: dict, tag: bool = True, _urlopen: Callable[[...], HTTPResponse] | None = None)

Uploads a manifest/index to a registry

Parameters:

• ref – The image reference.

• manifest – The manifest or index.

• tag – When true, use the tag, otherwise use the digest, this is relevant for multi-arch images, where the tag is an index, referencing the manifests by digest.

Returns:

The digest and size of the uploaded manifest.

spack.oci.oci.upload_manifest_with_retry(ref: ImageReference, manifest: dict, tag: bool = True, _urlopen: Callable[[...], HTTPResponse] | None = None)

Same as upload_manifest, but with retry wrapper

spack.oci.oci.with_query_param(url: str, param: str, value: str) → str

Add a query parameter to a URL

Parameters:

• url – The URL to add the parameter to.

• param – The parameter name.

• value – The parameter value.

Returns:

The URL with the parameter added.

spack.oci.opener module

All the logic for OCI fetching and authentication

class spack.oci.opener.Challenge(scheme: str | None = None, params: List[Tuple[str, str]] | None = None)

Bases: object

get_param(key: str) → str | None

Get the value of an auth param by key, or None if not found.

matches_scheme(scheme: str) → bool

Checks whether the challenge matches the given scheme, case-insensitive.

params

scheme

class spack.oci.opener.OCIAuthHandler(credentials_provider: Callable[[str], UsernamePassword | None])

Bases: BaseHandler

cached_auth_headers: Dict[str, str]

http_error_401(req: Request, fp, code, msg, headers)

https_request(req: Request)

class spack.oci.opener.RealmServiceScope(realm, service, scope)

Bases: NamedTuple

realm: str

Alias for field number 0

scope: str

Alias for field number 2

service: str

Alias for field number 1

class spack.oci.opener.State(*values)

Bases: Enum

AUTH_PARAM

AUTH_PARAM_LIST_START

AUTH_PARAM_OR_SCHEME

CHALLENGE

NEXT_IN_LIST

class spack.oci.opener.UsernamePassword(username, password)

Bases: NamedTuple

property basic_auth_header: str

password: str

Alias for field number 1

username: str

Alias for field number 0

class spack.oci.opener.WwwAuthenticateTokens(*values)

Bases: TokenBase

ANY

AUTH_PARAM

COMMA

EOF

EQUALS

SPACE

TOKEN

spack.oci.opener.create_opener()

Create an opener that can handle OCI authentication.

spack.oci.opener.credentials_from_mirrors(domain: str, *, mirrors: Iterable[Mirror] | None = None) → UsernamePassword | None

Filter out OCI registry credentials from a list of mirrors.

spack.oci.opener.ensure_status(request: Request, response: HTTPResponse, status: int)

Raise an error if the response status is not the expected one.

spack.oci.opener.parse_www_authenticate(input: str)

Very basic parsing of www-authenticate parsing (RFC7235 section 4.1) Notice: this omits token68 support.

spack.oci.opener.urlopen: Callable[[...], HTTPResponse]

Opener that automatically uses OCI authentication based on mirror config