Cosign sign manifest list only [CLOUDDST-24102]

-r is added to cosign sign in pubtools-sign which signs both manifest
list and each arch image. Only pass manifest list to cosign signer.

It also fixed manifest list digest in tag docker.

src/pubtools/_quay/iib_operations.py

@@ -122,7 +122,7 @@ def _index_image_to_sign_entries(
         if m["platform"]["architecture"] in ["amd64", "x86_64"]
     ]
     to_sign_entries = []
-    # if signing external images, sign also manifest list
+    # if signing external images, sign manifest list only
     if internal:
         for _dest_tag in dest_tags:
             for registry in dest_registries:
@@ -137,14 +137,12 @@ def _index_image_to_sign_entries(
                             signing_key=key,
                         )
                     )
+        return to_sign_entries
 
     for registry in dest_registries:
         for _dest_tag in dest_tags:
             for digest in index_image_digests:
-                if internal:
-                    reference = f"quay.io/{iib_repo}:{_dest_tag}"
-                else:
-                    reference = f"{registry}/{iib_repo}:{_dest_tag}"
+                reference = f"{registry}/{iib_repo}:{_dest_tag}"
                 for key in signing_keys:
                     to_sign_entries.append(
                         SignEntry(

src/pubtools/_quay/item_processor.py

@@ -479,13 +479,12 @@ def generate_repo_dest_tag_map(self, item: PushItem) -> Dict[str, Dict[str, List
                     ret.setdefault(registry, {}).setdefault(repo, []).append(tag)
         return ret
 
-    def generate_to_sign(self, item: Any, include_manifest_lists: bool = False) -> List[SignEntry]:
+    def generate_to_sign(self, item: Any, full_extract_ml: bool = True) -> List[SignEntry]:
         """Generate list of images to sign.
 
         Args:
             item (PushItem): Push item.
-            sign_only_arches(List[str]): List of architectures to sign.
-            If empty, all sign architectures.
+            full_extract_ml (bool): Extract manifest list or not.
         Returns:
             list: List of dictionaries containing reference, digest, repository and architecture.
         """
@@ -495,15 +494,22 @@ def generate_to_sign(self, item: Any, include_manifest_lists: bool = False) -> L
         )
 
         full_extract_orig = self.extractor.full_extract
-        self.extractor.full_extract = True
+        if full_extract_ml:
+            self.extractor.full_extract = True
         man_arch_digs = self.extractor.extract_manifests(item.metadata["pull_url"], media_types)
         self.extractor.full_extract = full_extract_orig
 
         for registry, repo, tag in self.generate_repo_dest_tags(item):
             reference = self.reference_processor.full_reference(registry, repo, tag)
+            if full_extract_ml:
+                man_arch_digs = [
+                    m for m in man_arch_digs if m.type_ != QuayClient.MANIFEST_LIST_TYPE
+                ]
+            elif any(m.type_ == QuayClient.MANIFEST_LIST_TYPE for m in man_arch_digs):
+                man_arch_digs = [
+                    m for m in man_arch_digs if m.type_ == QuayClient.MANIFEST_LIST_TYPE
+                ]
             for mad in man_arch_digs:
-                if mad.type_ == QuayClient.MANIFEST_LIST_TYPE and not include_manifest_lists:
-                    continue
                 public_registries = self.public_registries if registry == "quay.io" else [registry]
                 for public_registry in public_registries:
                     to_sign.append(

src/pubtools/_quay/push_docker.py

@@ -545,33 +545,12 @@ def sign_new_manifests(self, docker_push_items: List[Any]) -> List[Tuple[str, st
         current_signatures = []
         to_sign_new_entries = self.fetch_missing_push_items_digests(docker_push_items)
         to_sign_entries = []
-        to_sign_entries_internal = []
-        for internal_reg, repo_tags in to_sign_new_entries.items():
+        for _, repo_tags in to_sign_new_entries.items():
             for repo, tag_digests in repo_tags.items():
                 for tag, digests in tag_digests.items():
                     for type_, digest_key in digests.items():
                         digest, key = digest_key
-                        internal_reference = (
-                            f"{internal_reg}/"
-                            + self.target_settings["quay_namespace"]
-                            + "/"
-                            + get_internal_container_repo_name(repo)
-                            + ":"
-                            + tag
-                        )
                         for registry in self.dest_registries:
-                            pub_reference = f"{registry}/{repo}:{tag}"
-                            # add entries in internal format for cosign
-                            to_sign_entries_internal.append(
-                                SignEntry(
-                                    reference=internal_reference,
-                                    pub_reference=pub_reference,
-                                    repo=repo,
-                                    digest=digest,
-                                    signing_key=key,
-                                    arch="amd64",
-                                )
-                            )
                             reference = f"{registry}/{repo}:{tag}"
                             to_sign_entries.append(
                                 SignEntry(
@@ -591,8 +570,6 @@ def sign_new_manifests(self, docker_push_items: List[Any]) -> List[Tuple[str, st
                 signer = signercls(config_file=signer["config_file"], settings=self.target_settings)
                 if signercls.pre_push is True:
                     signer.sign_containers(to_sign_entries, self.task_id)
-                else:
-                    signer.sign_containers(to_sign_entries_internal, self.task_id)
 
         return current_signatures
 
@@ -678,10 +655,7 @@ def run(self) -> None:
         )
         to_sign_map = run_in_parallel(
             item_processor.generate_to_sign,
-            [
-                FData(args=(item,), kwargs={"include_manifest_lists": True})
-                for item in docker_push_items
-            ],
+            [FData(args=(item,), kwargs={"full_extract_ml": False}) for item in docker_push_items],
         )
         for _to_sign_entries in to_sign_map.values():
             to_sign_entries.extend(_to_sign_entries)

src/pubtools/_quay/tag_docker.py

@@ -15,7 +15,6 @@
 )
 from .utils.misc import (
     get_internal_container_repo_name,
-    get_pyxis_ssl_paths,
     set_aws_kms_environment_variables,
 )
 from .quay_client import QuayClient
@@ -510,8 +509,6 @@ def copy_tag_sign_images(self, push_item: Any, tag: str, executor: Executor) ->
                 Push item to perform the workflow with.
             tag (str):
                 Tag, which acts as a destination to the copy operation.
-            signature_handler (BasicSignatureHandler):
-                Instance of signature handler which will perform the signing.
             executor (Executor):
                 Instance of Executor subclass used for skopeo inspect.
         """
@@ -573,7 +570,6 @@ def copy_tag_sign_images(self, push_item: Any, tag: str, executor: Executor) ->
                     )
                 )
 
-            cert, key = get_pyxis_ssl_paths(self.target_settings)
             item_processor = item_processor_for_internal_data(
                 self.quay_client,
                 self.target_settings["quay_host"].rstrip("/"),
@@ -634,8 +630,6 @@ def merge_manifest_lists_sign_images(
                 Tag, which acts as a destination to the merge operation.
             add_archs ([str]):
                 Architectures which should be copied to the existing manifest list.
-            signature_handler (BasicSignatureHandler):
-                Instance of signature handler which will perform the signing.
         """
         LOG.info(
             "Architectures {0} of tag '{1}' will be copied to destination tag '{2}'".format(
@@ -669,21 +663,6 @@ def merge_manifest_lists_sign_images(
 
             for manifest in new_manifest_list["manifests"]:
                 for registry in dest_registries:
-                    to_sign_entries_internal.append(
-                        SignEntry(
-                            repo=repo,
-                            pub_reference=f"{registry}/{repo}:{tag}",
-                            reference="quay.io/"
-                            + self.target_settings["quay_namespace"]
-                            + "/"
-                            + internal_repo
-                            + ":"
-                            + tag,
-                            digest=manifest["digest"],
-                            arch=manifest["platform"]["architecture"],
-                            signing_key=push_item.claims_signing_key,
-                        )
-                    )
                     reference = external_image_schema.format(host=registry, repo=repo, tag=tag)
                     to_sign_entries.append(
                         SignEntry(
@@ -699,7 +678,6 @@ def merge_manifest_lists_sign_images(
                         (reference, manifest["digest"], push_item.claims_signing_key)
                     )
 
-            namespace = self.target_settings["quay_namespace"]
             item_processor = item_processor_for_internal_data(
                 self.quay_client,
                 self.target_settings["quay_host"].rstrip("/"),
@@ -748,11 +726,11 @@ def merge_manifest_lists_sign_images(
             ml_to_sign = raw_src_manifest
             self.quay_client.upload_manifest(raw_src_manifest, dest_image, raw=True)
         else:
-            ml_to_sign = json.dumps(new_manifest_list)
+            ml_to_sign = json.dumps(new_manifest_list, sort_keys=True, indent=4)
             self.quay_client.upload_manifest(new_manifest_list, dest_image)
 
         if push_item.claims_signing_key:
-            # for cosign sign also manifest list
+            # for cosign sign manifest list only
             digest = "sha256:" + hashlib.sha256(ml_to_sign.encode("utf-8")).hexdigest()
             for pub_registry in dest_registries:
                 to_sign_entries_internal.append(
@@ -837,8 +815,6 @@ def untag_image(self, push_item: Any, tag: str) -> None:
         )
         dest_image = "{0}:{1}".format(full_repo, tag)
 
-        cert, key = get_pyxis_ssl_paths(self.target_settings)
-
         for signer in self.target_settings["signing"]:
             if signer["enabled"]:
                 signercls = SIGNER_BY_LABEL[signer["label"]]