image_digest.resolve honours a @digest: asks the registry for THAT manifest (R-746)

The digest was dropped and the tag's current digest returned, so 'is this digest still served' got a false yes
(measured: redis:7-alpine@sha256:000…0 resolved to the tag's digest; now HTTP 404). A malformed digest is refused
without a request. test_image_digest.py (no network); red-proof: the pre-fix resolver fails 3 of 4 cases.

Co-Authored-By: Claude Opus 5.5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0159rPz1ZhFKsS53msqPYxtS

scripts/image_digest.py

@@ -34,8 +34,8 @@ UA = "felhom-catalog-digest/1.0 (read-only)"
 
 
 def split_ref(ref):
-    """'ghcr.io/a/b:1.2' -> ('ghcr.io', 'a/b', '1.2'). A digest suffix is dropped; Docker Hub
-    short names get `library/`."""
+    """'ghcr.io/a/b:1.2' -> ('ghcr.io', 'a/b', '1.2'). A digest suffix is dropped here (resolve() asks for it by
+    itself); Docker Hub short names get `library/`."""
     ref = ref.split("@", 1)[0]
     first = ref.split("/", 1)[0]
     if "/" in ref and ("." in first or ":" in first or first == "localhost"):
@@ -69,15 +69,23 @@ def _bearer(www_auth):
 
 
 def resolve(ref, timeout=30):
-    """(digest, None) or (None, why). Read-only: one HEAD, one token fetch at most."""
+    """(digest, None) or (None, why). Read-only: one HEAD, one token fetch at most.
+
+    A ref carrying `@sha256:…` asks for THAT manifest (R-746): the answer is the digest only when the registry still
+    serves it, never the tag's current digest — a caller asking "is this digest still served" got a false yes."""
     host, repo, tag = split_ref(ref)
-    url = "https://%s/v2/%s/manifests/%s" % (host, repo, tag)
+    want = ref.split("@", 1)[1] if "@" in ref else ""
+    if want and not (want.startswith("sha256:") and len(want) == 71):
+        return None, "not a sha256 digest: %s" % want
+    url = "https://%s/v2/%s/manifests/%s" % (host, repo, want or tag)
     headers = {"Accept": ACCEPT, "User-Agent": UA}
     for attempt in (1, 2):
         req = urllib.request.Request(url, headers=headers, method="HEAD")
         try:
             with urllib.request.urlopen(req, timeout=timeout) as r:
                 d = r.headers.get("Docker-Content-Digest")
+                if want and d != want:
+                    return None, "the registry answered %s for %s" % (d, want)
                 if d and d.startswith("sha256:") and len(d) == 71:
                     return d, None
                 return None, "no Docker-Content-Digest header (HTTP %s)" % r.status

scripts/test_image_digest.py

@@ -0,0 +1,75 @@
+#!/usr/bin/env python3
+"""R-746: image_digest.resolve honours a `@digest` — it asks the registry for THAT manifest and never answers the tag's
+current digest. No network: urlopen is replaced by a fake registry that serves one tag and one digest.
+
+COMPANION RED-PROOF: with the pre-fix resolve (it dropped the digest and asked for the tag), the absent-digest case
+answers the tag's digest and this test fails ("a digest the registry does not serve was resolved")."""
+import io
+import os
+import sys
+import unittest
+import urllib.error
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+import image_digest  # noqa: E402
+
+SERVED = "sha256:" + "a" * 64
+GONE = "sha256:" + "0" * 64
+
+
+class _Resp:
+    def __init__(self, digest):
+        self.headers = {"Docker-Content-Digest": digest}
+        self.status = 200
+
+    def __enter__(self):
+        return self
+
+    def __exit__(self, *a):
+        return False
+
+
+class FakeRegistry:
+    def __init__(self):
+        self.urls = []
+
+    def __call__(self, req, timeout=30):
+        url = req.full_url
+        self.urls.append(url)
+        last = url.rsplit("/", 1)[-1]
+        if last in ("7-alpine", SERVED):  # the tag and the digest it serves
+            return _Resp(SERVED)
+        raise urllib.error.HTTPError(url, 404, "not found", {}, io.BytesIO(b""))
+
+
+class ResolveDigest(unittest.TestCase):
+    def setUp(self):
+        self.fake = FakeRegistry()
+        self.orig = image_digest.urllib.request.urlopen
+        image_digest.urllib.request.urlopen = self.fake
+
+    def tearDown(self):
+        image_digest.urllib.request.urlopen = self.orig
+
+    def test_plain_tag_unchanged(self):
+        self.assertEqual(image_digest.resolve("redis:7-alpine"), (SERVED, None))
+        self.assertTrue(self.fake.urls[-1].endswith("/manifests/7-alpine"))
+
+    def test_served_digest_is_asked_by_digest(self):
+        d, why = image_digest.resolve("redis:7-alpine@" + SERVED)
+        self.assertEqual(d, SERVED, why)
+        self.assertTrue(self.fake.urls[-1].endswith("/manifests/" + SERVED), self.fake.urls)
+
+    def test_absent_digest_is_refused(self):
+        d, why = image_digest.resolve("redis:7-alpine@" + GONE)
+        self.assertIsNone(d, "a digest the registry does not serve was resolved (to %s)" % d)
+        self.assertIn("404", why)
+
+    def test_malformed_digest_is_refused(self):
+        d, why = image_digest.resolve("redis:7-alpine@sha256:abc")
+        self.assertIsNone(d)
+        self.assertEqual(self.fake.urls, [], "a malformed digest must not reach the registry")
+
+
+if __name__ == "__main__":
+    unittest.main()