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)

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) → ImageReference

manifest_url() → str

name

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_index_tag = 'index.spack'

Default OCI index tag

spack.oci.image.default_manifest()

spack.oci.image.default_tag(spec: Spec) → str

Return a valid, default image tag for a spec.

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.tag_is_spec(tag: str) → bool

Check if a tag is likely a Spec

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 = ['application/vnd.oci.image.manifest.v1+json', 'application/vnd.docker.distribution.manifest.v2+json', 'application/vnd.oci.image.index.v1+json', 'application/vnd.docker.distribution.manifest.list.v2+json']

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(*args, **kwargs)

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(*args, **kwargs)

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 = ['application/vnd.oci.image.index.v1+json', 'application/vnd.docker.distribution.manifest.list.v2+json']

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 = ['application/vnd.oci.image.manifest.v1+json', 'application/vnd.docker.distribution.manifest.v2+json']

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(*args, **kwargs)

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(*args, **kwargs)

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

params

scheme

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

Bases: BaseHandler

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

https_request(req: Request)

obtain_bearer_token(registry: str, challenge: RealmServiceScope, timeout) → str

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(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Enum

AUTH_PARAM = 3

AUTH_PARAM_LIST_START = 2

AUTH_PARAM_OR_SCHEME = 5

CHALLENGE = 1

NEXT_IN_LIST = 4

class spack.oci.opener.TokenType(value, names=None, *, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: TokenBase

ANY = 7

AUTH_PARAM = 1

COMMA = 4

EOF = 6

EQUALS = 3

SPACE = 5

TOKEN = 2

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

Bases: NamedTuple

password: str

Alias for field number 1

username: str

Alias for field number 0

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.default_retry(f, retries: int = 5, sleep=None)

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.get_bearer_challenge(challenges: List[Challenge]) → RealmServiceScope | None

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.tokenize(input: str)

spack.oci.opener.urlopen: Callable[[...], HTTPResponse] = <function _urlopen.<locals>.dispatch_open>

Opener that automatically uses OCI authentication based on mirror config