From 8a07481a463b4a4ca5e08660c322d8143e95a644 Mon Sep 17 00:00:00 2001
From: Robert Cerven <rcerven@redhat.com>
Date: Thu, 23 May 2024 18:44:05 +0200
Subject: [PATCH] Do not delete sig/att/sbom if subject image exists

when getting all tags in registry we are listing only active tags,
if tag was removed for the image, image still will be in registry
if it is part of image index

if manifest digest isn't present in any of active tags,
try query also manifest digest to verify its presence

STONEBLD-2466

Signed-off-by: Robert Cerven <rcerven@redhat.com>
---
 .../image_pruner/prune_images.py              | 34 ++++++++++++++++---
 1 file changed, 29 insertions(+), 5 deletions(-)

diff --git a/config/registry_image_pruner/image_pruner/prune_images.py b/config/registry_image_pruner/image_pruner/prune_images.py
index c2b4da9..0f3e0d5 100644
--- a/config/registry_image_pruner/image_pruner/prune_images.py
+++ b/config/registry_image_pruner/image_pruner/prune_images.py
@@ -79,6 +79,27 @@ def delete_image_tag(quay_token: str, namespace: str, name: str, tag: str) -> No
             raise(ex)
 
 
+def manifest_exists(quay_token: str, namespace: str, name: str, manifest: str) -> bool:
+    api_url = f"{QUAY_API_URL}/repository/{namespace}/{name}/manifest/{manifest}"
+    request = Request(api_url, headers={
+        "Authorization": f"Bearer {quay_token}",
+    })
+    resp: HTTPResponse
+    manifest_exists = True
+    try:
+        with urlopen(request) as resp:
+            if resp.status != 200 and resp.status != 204:
+                raise RuntimeError(resp.reason)
+
+    except HTTPError as ex:
+        if ex.status != 404:
+            raise(ex)
+        else:
+            manifest_exists = False
+
+    return manifest_exists
+
+
 def remove_tags(tags: List[Dict[str, Any]], quay_token: str, namespace: str, name: str, dry_run: bool = False) -> None:
     image_digests = [image["manifest_digest"] for image in tags]
     tags_map = {tag_info["name"]: tag_info for tag_info in tags}
@@ -87,11 +108,14 @@ def remove_tags(tags: List[Dict[str, Any]], quay_token: str, namespace: str, nam
         # attestation or sbom image
         if (match := tag_regex.match(tag["name"])) is not None:
             if f"sha256:{match.group(1)}" not in image_digests:
-                if dry_run:
-                    LOGGER.info("Tag %s from %s/%s should be removed", tag["name"], namespace, name)
-                else:
-                    LOGGER.info("Removing tag %s from %s/%s", tag["name"], namespace, name)
-                    delete_image_tag(quay_token, namespace, name, tag["name"])
+                # verify that manifest really doesn't exist, because if tag was removed, it won't be tag list, but still be in the registry
+                if not manifest_exists(quay_token, namespace, name, f"sha256:{match.group(1)}"):
+                    if dry_run:
+                        LOGGER.info("Tag %s from %s/%s should be removed", tag["name"], namespace, name)
+                    else:
+                        LOGGER.info("Removing tag %s from %s/%s", tag["name"], namespace, name)
+                        delete_image_tag(quay_token, namespace, name, tag["name"])
+
         elif tag["name"].endswith(".src"):
             to_delete = False