+ 

+     # Resolve container tags in extra_files

+     tag_resolver = util.ContainerTagResolver(offline=offline)

+     if config.get("extra_files"):

+         for _, arch_dict in config["extra_files"]:

+             for value in arch_dict.values():

+                 if isinstance(value, dict):

+                     _resolve_container_tag(value, tag_resolver)

+                 elif isinstance(value, list):

+                     for subinstance in value:

+                         _resolve_container_tag(subinstance, tag_resolver)

+     if config.get("extra_isos"):

+         for cfgs in config["extra_isos"].values():

+             if not isinstance(cfgs, list):

+                 cfgs = [cfgs]

+             for cfg in cfgs:

+                 if isinstance(cfg.get("extra_files"), dict):

+                     _resolve_container_tag(cfg["extra_files"], tag_resolver)

+                 elif isinstance(cfg.get("extra_files"), list):

+                     for c in cfg["extra_files"]:

+                         _resolve_container_tag(c, tag_resolver)

+ 

+             "extra_file": {

+                 "type": "object",

+                 "properties": {

+                     "scm": {"type": "string"},

+                     "repo": {"type": "string"},

+                     "branch": {"$ref": "#/definitions/optional_string"},

+                     "file": {"$ref": "#/definitions/strings"},

+                     "dir": {"$ref": "#/definitions/strings"},

+                     "target": {"type": "string"},

+                 },

+                 "additionalProperties": False,

+             },

+                                     {"$ref": "#/definitions/extra_file"}

+                 {"type": "array", "items": {"$ref": "#/definitions/extra_file"}}

+ 

+ 

+ def _resolve_container_tag(instance, tag_resolver):

+     if instance.get("scm") == "container-image":

+         instance["repo"] = tag_resolver(instance["repo"])

+ class ContainerTagResolver(object):

+     """

+     A caching resolver for container image urls that replaces tags with digests.

+     """

+ 

+     def __init__(self, offline=False):

+         self.offline = offline

+         self.cache = {}

+ 

+     def __call__(self, url):

+         if self.offline:

+             # We're offline, nothing to do

+             return url

+         if re.match(".*@sha256:[a-z0.9]+", url):

+             # We already have a digest

+             return url

+         if url not in self.cache:

+             self.cache[url] = self._resolve(url)

+         return self.cache[url]

+ 

+     def _resolve(self, url):

+         m = re.match("^.+(:.+)$", url)

+         if not m:

+             raise RuntimeError("Failed to find tag name")

+         tag = m.group(1)

+ 

+         data = _skopeo_inspect(url)

+         digest = data["Digest"]

+         return url.replace(tag, f"@{digest}")

+ 

+ 

+ def _skopeo_inspect(url):

+     """Wrapper for running `skopeo inspect {url}` and parsing the output."""

+     cp = subprocess.run(

+         ["skopeo", "inspect", url], stdout=subprocess.PIPE, check=True, encoding="utf-8"

+     )

+     return json.loads(cp.stdout)

+ 

+ 

+ class TestContainerTagResolver(unittest.TestCase):

+     @mock.patch("pungi.util._skopeo_inspect")

+     def test_offline(self, inspect):

+         resolver = util.ContainerTagResolver(offline=True)

+         url = "docker://example.com/repo:latest"

+         assert url == resolver(url)

+         assert inspect.mock_calls == []

+ 

+     @mock.patch("pungi.util._skopeo_inspect")

+     def test_already_digest(self, inspect):

+         resolver = util.ContainerTagResolver()

+         url = "docker://example.com/repo@sha256:abcdef0123456789"

+         assert url == resolver(url)

+         assert inspect.mock_calls == []

+ 

+     @mock.patch("pungi.util._skopeo_inspect")

+     def test_simple(self, inspect):

+         url = "docker://example.com/repo"

+         digest = "sha256:abcdef"

+         orig_url = f"{url}:latest"

+         inspect.return_value = {"Digest": digest}

+         resolver = util.ContainerTagResolver()

+         assert f"{url}@{digest}" == resolver(orig_url)

+         assert inspect.mock_calls == [mock.call(orig_url)]

+ 

+     @mock.patch("pungi.util._skopeo_inspect")

+     def test_caching(self, inspect):

+         url = "docker://example.com/repo"

+         digest = "sha256:abcdef"

+         orig_url = f"{url}:latest"

+         inspect.return_value = {"Digest": digest}

+         resolver = util.ContainerTagResolver()

+         assert f"{url}@{digest}" == resolver(orig_url)

+         assert f"{url}@{digest}" == resolver(orig_url)

+         assert f"{url}@{digest}" == resolver(orig_url)

+         assert inspect.mock_calls == [mock.call(orig_url)]

+ 

+