diff --git a/scripts/image_digest.py b/scripts/image_digest.py index be8dd33..c497f7f 100644 --- a/scripts/image_digest.py +++ b/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 diff --git a/scripts/test_image_digest.py b/scripts/test_image_digest.py new file mode 100644 index 0000000..879a996 --- /dev/null +++ b/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()