6ba0561187
storage_conf was never not None, so the loading was called every time. This never crashed because conf was always being set, but it wasn't working properly regardless.
104 lines
3.7 KiB
Python
Executable file
104 lines
3.7 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)
|
|
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()
|