cde6e1e114
An error message has changed in skopeo [1] (or one of it's underlying
libraries). This new version is now in our CI so let's fix our tests.
[1]: f423f01d1b
Signed-off-by: Simon de Vlieger <supakeen@redhat.com>
105 lines
3.8 KiB
Python
Executable file
105 lines
3.8 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 hashlib
|
|
import subprocess as sp
|
|
import sys
|
|
|
|
from osbuild import sources
|
|
from osbuild.util import host
|
|
|
|
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:
|
|
self.storage_conf = host.get_container_storage()
|
|
driver = self.storage_conf["storage"]["driver"]
|
|
graphroot = self.storage_conf["storage"]["graphroot"]
|
|
runroot = self.storage_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:
|
|
for checksum in items:
|
|
self.fetch_one(checksum, None)
|
|
|
|
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)
|
|
# update: 2025, the message did change!
|
|
if "identifier is not an image" in res.stderr or "does not resolve to an image ID" 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()
|