82f2414637
When an images does not exist just return `False` instead of raising a RuntimeError. If anything else goes wrong (unknown output or hash mismatch) keep the RuntimeError as this is an unexpected exception.
108 lines
3.9 KiB
Python
Executable file
108 lines
3.9 KiB
Python
Executable file
#!/usr/bin/python3
|
|
"""Provide access to an image in the host system's container storage.
|
|
|
|
This stage differs from other sources in that the source checks to see if the
|
|
container is available in the host's container storage. This puts a requirement
|
|
on the user to ensure that the container is copied into local storage before
|
|
trying to build an image. The starts by reading the host's `storage.conf`
|
|
file and then using the config to check if the container has been imported.
|
|
|
|
Unlike all other sources, this source relies on external storage not controlled
|
|
by osbuild itself.
|
|
|
|
Buildhost commands used: `skopeo`.
|
|
"""
|
|
|
|
import concurrent.futures
|
|
import hashlib
|
|
import subprocess as sp
|
|
import sys
|
|
|
|
from osbuild import sources
|
|
from osbuild.util import containers
|
|
|
|
SCHEMA = """
|
|
"additionalProperties": false,
|
|
"properties": {
|
|
"items": {
|
|
"description": "The container image to fetch indexed by the container image id",
|
|
"type": "object",
|
|
"additionalProperties": false,
|
|
"patternProperties": {
|
|
"sha256:[0-9a-f]{64}": {
|
|
"type": "object",
|
|
"additionalProperties": false
|
|
}
|
|
}
|
|
}
|
|
}
|
|
"""
|
|
|
|
|
|
class ContainersStorageSource(sources.SourceService):
|
|
|
|
content_type = "org.osbuild.containers-storage"
|
|
|
|
storage_conf = None
|
|
|
|
def local_image_name(self, imagename):
|
|
"""
|
|
Construct the full image name that references an image with a given checksum in the local storage.
|
|
"""
|
|
if self.storage_conf is None:
|
|
conf = containers.get_host_storage()
|
|
driver = conf["storage"]["driver"]
|
|
graphroot = conf["storage"]["graphroot"]
|
|
runroot = conf["storage"]["runroot"]
|
|
|
|
return f"containers-storage:[{driver}@{graphroot}+{runroot}]{imagename}"
|
|
|
|
def fetch_one(self, checksum, desc) -> None:
|
|
# Instead of fetching anything, just check that it exists.
|
|
#
|
|
# Note that there's an obvious TOCTOU issue here, but it's unavoidable without copying the storage or a
|
|
# container out of it, which is exactly what we want to avoid with this source.
|
|
# Unlike all other sources, this source relies on external storage not controlled by osbuild itself.
|
|
self.exists(checksum, desc)
|
|
|
|
def fetch_all(self, items) -> None:
|
|
# prepare each item as a (checksum, desc) tuple (where desc=None)
|
|
transformed = map(lambda i: self.transform(i, None), items)
|
|
with concurrent.futures.ThreadPoolExecutor(max_workers=self.max_workers) as executor:
|
|
for _ in executor.map(self.fetch_one, *zip(*transformed)):
|
|
pass
|
|
|
|
def exists(self, checksum, _) -> bool:
|
|
image_id = checksum.split(":")[1]
|
|
source = self.local_image_name(image_id)
|
|
res = sp.run(["skopeo", "inspect", "--raw", "--config", source],
|
|
check=False, stdout=sp.PIPE, stderr=sp.PIPE, universal_newlines=True)
|
|
|
|
# fail early if the user hasn't imported the container into
|
|
# containers-storage
|
|
if res.returncode != 0:
|
|
# string not matching not ideal - this is ErrNotAnImage
|
|
# which is unchanged since 2016 (added in ee99172905 in
|
|
# containers/storage)
|
|
if "identifier is not an image" in res.stderr:
|
|
return False
|
|
raise RuntimeError(f"unknown skopeo error: {res.stderr}")
|
|
|
|
# NOTE: this is a bit redundant because we're checking the content digest of the thing we retrieved via its
|
|
# id (which is the content digest), but let's keep it in case we change to retrieving local containers by name
|
|
# See also https://github.com/containers/skopeo/pull/2236
|
|
local_id = "sha256:" + hashlib.sha256(res.stdout.encode()).hexdigest()
|
|
if local_id != checksum:
|
|
raise RuntimeError(
|
|
f"Local container image id of {local_id}, but expected {checksum}")
|
|
|
|
return True
|
|
|
|
|
|
def main():
|
|
service = ContainersStorageSource.from_args(sys.argv[1:])
|
|
service.main()
|
|
|
|
|
|
if __name__ == '__main__':
|
|
main()
|