From ea29e985da906ebb425b801e26a09d907485a366 Mon Sep 17 00:00:00 2001 From: parmstro Date: Thu, 12 Nov 2026 03:17:20 +0000 Subject: [PATCH] Skip /tags/list enumeration for digest-only sync When all entries in a remote's includes field are sha256 digests, bypass the expensive /tags/list enumeration and fetch manifests directly by digest. Cosign companion tags are discovered via cached HEAD probes filtered through the remote's excludes, fixing the signed_only ordering bug where signature checks ran before cosign discovery. Only HTTP 404 is treated as tag absence; auth failures and rate limits propagate. Tested against Satellite 6.19.5 with openshift-release-dev/ocp-v4.0-art-dev (50,000+ tags). SAT-49501 --- CHANGES/2474.feature | 1 + pulp_container/app/tasks/sync_stages.py | 95 ++++++- pulp_container/app/tasks/synchronize.py | 2 +- pulp_container/tests/unit/test_sync_stages.py | 242 +++++++++++++++++- 4 files changed, 336 insertions(+), 4 deletions(-) create mode 100644 CHANGES/2474.feature diff --git a/CHANGES/2474.feature b/CHANGES/2474.feature new file mode 100644 index 000000000..8f97ff9c1 --- /dev/null +++ b/CHANGES/2474.feature @@ -0,0 +1 @@ +Added tag list bypass optimization for digest-based sync. When all entries in a remote's `includes` field are sha256 digests, the sync pipeline skips the expensive `/tags/list` enumeration and fetches manifests directly by digest. diff --git a/pulp_container/app/tasks/sync_stages.py b/pulp_container/app/tasks/sync_stages.py index d24e62bb4..b46e0103e 100644 --- a/pulp_container/app/tasks/sync_stages.py +++ b/pulp_container/app/tasks/sync_stages.py @@ -3,6 +3,7 @@ import hashlib import json import logging +import re from collections import defaultdict from urllib.parse import urljoin, urlparse, urlunparse @@ -52,12 +53,13 @@ class ContainerFirstStage(Stage): """ - def __init__(self, remote, signed_only): + def __init__(self, remote, signed_only, mirror=False): """Initialize the stage.""" super().__init__() self.remote = remote self.deferred_download = self.remote.policy != Remote.IMMEDIATE self.signed_only = signed_only + self.mirror = mirror self.tag_dcs = [] self.manifest_list_dcs = [] @@ -66,6 +68,9 @@ def __init__(self, remote, signed_only): self._synced_digests = defaultdict(list) self._full_tag_list = [] self._cosign_tags = [] + self._bypass_taglist = False + self._probed_cosign_digests = set() + self._cosign_probe_semaphore = asyncio.Semaphore(20) async def _download_manifest_data(self, manifest_url): downloader = self.remote.get_downloader(url=manifest_url) @@ -116,12 +121,99 @@ async def _check_for_existing_manifest(self, head_manifest_task): return content_data, raw_text_data, response, original_reference + def _can_bypass_taglist(self): + """ + Check if we can safely bypass /tags/list enumeration. + + Returns True only if all includes are explicit sha256 digests + and we are not in mirror mode (which needs the full list to detect removals). + """ + includes = self.remote.includes or [] + + if not includes: + return False + + if self.mirror: + return False + + return all(re.fullmatch(r"sha256:[0-9a-f]{64}", entry) for entry in includes) + + async def _tag_exists(self, tag_name): + """Check if a tag exists upstream via lightweight HEAD request.""" + relative_url = "/v2/{name}/manifests/{tag}".format( + name=self.remote.namespaced_upstream_name, tag=tag_name + ) + manifest_url = urljoin(self.remote.url, relative_url) + downloader = self.remote.get_downloader(url=manifest_url) + + try: + await downloader.run(extra_data={"headers": V2_ACCEPT_HEADERS, "http_method": "head"}) + return True + except aiohttp.ClientResponseError as exc: + if exc.status == 404: + return False + raise + + async def _ensure_cosign_tags(self, digests): + """ + Populate _cosign_tags by probing for cosign companion tags via HEAD requests. + + Only used when _bypass_taglist is True. Results are cached per digest + so repeated calls (e.g. from _has_cosign_signature and then the post-sync + companion pass) do not re-probe. + """ + if not self._bypass_taglist: + return + + pending = sorted(set(digests) - self._probed_cosign_digests) + candidates = [] + + for digest in pending: + if not digest.startswith("sha256:"): + continue + + stem = digest.replace("sha256:", "sha256-", 1) + candidates.extend(f"{stem}{suffix}" for suffix in (*COSIGN_TAG_SUFFIXES, "")) + + candidates = filter_resources(candidates, [], self.remote.excludes or []) + + async def probe(tag): + async with self._cosign_probe_semaphore: + return tag if await self._tag_exists(tag) else None + + results = await asyncio.gather(*(probe(tag) for tag in candidates)) + self._cosign_tags.extend(tag for tag in results if tag is not None) + self._probed_cosign_digests.update(pending) + async def run(self): """ ContainerFirstStage. """ signature_source = await self.get_signature_source() + self._bypass_taglist = self._can_bypass_taglist() + if self._bypass_taglist: + digest_list = list(self.remote.includes) + log.info( + "Bypassing /tags/list enumeration - syncing %d explicit digests directly", + len(digest_list), + ) + await self._process_manifests(digest_list, signature_source, "Processing Manifests") + + await self._ensure_cosign_tags(list(self._synced_digests)) + companion_tags = self._find_cosign_companion_tags() + if companion_tags: + log.info( + "Found %d cosign companion tag(s) for synced manifests", + len(companion_tags), + ) + await self._process_manifests( + companion_tags, + signature_source, + "Processing Cosign Companion Tags", + ) + return + async with ProgressReport( message="Downloading tag list", code="sync.downloading.tag_list", total=1 ) as pb: @@ -300,6 +392,7 @@ async def _process_manifests(self, manifests, signature_source, msg): async def _has_cosign_signature(self, digest): """Check if a digest has a cosign signature.""" + await self._ensure_cosign_tags([digest]) cosign_digest = digest.replace("sha256:", "sha256-") if f"{cosign_digest}.sig" in self._cosign_tags: return True diff --git a/pulp_container/app/tasks/synchronize.py b/pulp_container/app/tasks/synchronize.py index aa2ffa051..9a3cad78d 100644 --- a/pulp_container/app/tasks/synchronize.py +++ b/pulp_container/app/tasks/synchronize.py @@ -37,7 +37,7 @@ def synchronize(remote_pk, repository_pk, mirror, signed_only): remote = ContainerRemote.objects.get(pk=remote_pk) repository = ContainerRepository.objects.get(pk=repository_pk) log.info("Synchronizing: repository={r} remote={p}".format(r=repository.name, p=remote.name)) - first_stage = ContainerFirstStage(remote, signed_only) + first_stage = ContainerFirstStage(remote, signed_only, mirror=mirror) dv = ContainerDeclarativeVersion(first_stage, repository, mirror) rv = dv.create() if rv: diff --git a/pulp_container/tests/unit/test_sync_stages.py b/pulp_container/tests/unit/test_sync_stages.py index 835347cb7..27b1af36d 100644 --- a/pulp_container/tests/unit/test_sync_stages.py +++ b/pulp_container/tests/unit/test_sync_stages.py @@ -1,8 +1,10 @@ -"""Unit tests for cosign companion tag helpers on ContainerFirstStage.""" +"""Unit tests for ContainerFirstStage: cosign companion tag helpers and tag list bypass.""" import unittest from unittest.mock import AsyncMock, MagicMock, patch +import aiohttp + from pulp_container.app.tasks.sync_stages import COSIGN_TAG_SUFFIXES, ContainerFirstStage from pulp_container.constants import MEDIA_TYPE @@ -98,8 +100,16 @@ async def test_has_cosign_signature_true_when_sig_tag_present(self): cosign_key = digest.replace("sha256:", "sha256-") self.stage._cosign_tags = [f"{cosign_key}.sig"] + self.stage._ensure_cosign_tags = AsyncMock() self.assertTrue(await self.stage._has_cosign_signature(digest)) - self.stage.remote.get_downloader.assert_not_called() + + async def test_has_cosign_signature_calls_ensure_cosign_tags(self): + _, digest = _bare_cosign_digest() + self.stage._cosign_tags = [] + self.stage._ensure_cosign_tags = AsyncMock() + + await self.stage._has_cosign_signature(digest) + self.stage._ensure_cosign_tags.assert_awaited_once_with([digest]) async def test_has_cosign_signature_true_after_fetching_v3_index(self): tag, digest = _bare_cosign_digest() @@ -115,6 +125,7 @@ async def test_has_cosign_signature_true_after_fetching_v3_index(self): mock_response = MagicMock() mock_response.url = f"https://registry.example/v2/foo/manifests/{tag}" + self.stage._ensure_cosign_tags = AsyncMock() self.stage._download_manifest_data = AsyncMock( return_value=(content_data, raw, mock_response) ) @@ -136,6 +147,7 @@ async def test_has_cosign_signature_false_when_bare_tag_not_companion(self): mock_response = MagicMock() mock_response.url = f"https://registry.example/v2/foo/manifests/{tag}" + self.stage._ensure_cosign_tags = AsyncMock() self.stage._download_manifest_data = AsyncMock( return_value=(content_data, raw, mock_response) ) @@ -149,8 +161,234 @@ async def test_has_cosign_signature_false_when_bare_tag_not_companion(self): async def test_has_cosign_signature_false_when_no_cosign_tags(self): _, digest = _bare_cosign_digest() self.stage._cosign_tags = [] + self.stage._ensure_cosign_tags = AsyncMock() self.assertFalse(await self.stage._has_cosign_signature(digest)) +class TestCanBypassTaglist(unittest.TestCase): + """Test _can_bypass_taglist decision logic.""" + + def _make_stage(self, includes=None, excludes=None, mirror=False): + remote = MagicMock() + remote.policy = MagicMock() + remote.includes = includes + remote.excludes = excludes + stage = ContainerFirstStage(remote=remote, signed_only=False, mirror=mirror) + return stage + + def test_all_digests_no_excludes(self): + stage = self._make_stage(includes=["sha256:" + "a" * 64, "sha256:" + "b" * 64]) + assert stage._can_bypass_taglist() is True + + def test_non_digest_entry_prevents_bypass(self): + stage = self._make_stage(includes=["latest", "sha256:" + "a" * 64]) + assert stage._can_bypass_taglist() is False + + def test_alias_entry_prevents_bypass(self): + stage = self._make_stage(includes=["cli=sha256:" + "a" * 64]) + assert stage._can_bypass_taglist() is False + + def test_wildcard_entry_prevents_bypass(self): + stage = self._make_stage(includes=["v1.*"]) + assert stage._can_bypass_taglist() is False + + def test_empty_includes_prevents_bypass(self): + stage = self._make_stage(includes=[]) + assert stage._can_bypass_taglist() is False + + def test_none_includes_prevents_bypass(self): + stage = self._make_stage(includes=None) + assert stage._can_bypass_taglist() is False + + def test_mirror_mode_prevents_bypass(self): + stage = self._make_stage(includes=["sha256:" + "a" * 64], mirror=True) + assert stage._can_bypass_taglist() is False + + def test_uppercase_hex_prevents_bypass(self): + stage = self._make_stage(includes=["sha256:" + "A" * 64]) + assert stage._can_bypass_taglist() is False + + def test_short_hex_prevents_bypass(self): + stage = self._make_stage(includes=["sha256:" + "a" * 63]) + assert stage._can_bypass_taglist() is False + + def test_excludes_do_not_affect_bypass_decision(self): + stage = self._make_stage( + includes=["sha256:" + "a" * 64], + excludes=["sha256-*"], + ) + assert stage._can_bypass_taglist() is True + + +class TestEnsureCosignTags(unittest.IsolatedAsyncioTestCase): + """Test _ensure_cosign_tags cached HEAD-probe logic.""" + + def setUp(self): + remote = MagicMock() + remote.policy = MagicMock() + remote.namespaced_upstream_name = "library/test" + remote.url = "https://registry.example/" + remote.get_downloader = MagicMock() + remote.excludes = None + self.stage = ContainerFirstStage(remote=remote, signed_only=False) + self.stage._bypass_taglist = True + + async def test_probes_all_cosign_patterns(self): + digest = "sha256:" + "a" * 64 + hex_part = "a" * 64 + + self.stage._tag_exists = AsyncMock(return_value=True) + await self.stage._ensure_cosign_tags([digest]) + + expected_tags = { + f"sha256-{hex_part}.sig", + f"sha256-{hex_part}.att", + f"sha256-{hex_part}.sbom", + f"sha256-{hex_part}", + } + assert set(self.stage._cosign_tags) == expected_tags + + async def test_returns_only_existing_tags(self): + digest = "sha256:" + "a" * 64 + hex_part = "a" * 64 + + async def selective_exists(tag): + return tag.endswith(".sig") + + self.stage._tag_exists = selective_exists + await self.stage._ensure_cosign_tags([digest]) + + assert self.stage._cosign_tags == [f"sha256-{hex_part}.sig"] + + async def test_skips_non_sha256_digests(self): + self.stage._tag_exists = AsyncMock(return_value=True) + await self.stage._ensure_cosign_tags(["not-a-digest"]) + assert self.stage._cosign_tags == [] + self.stage._tag_exists.assert_not_called() + + async def test_empty_input(self): + self.stage._tag_exists = AsyncMock(return_value=True) + await self.stage._ensure_cosign_tags([]) + assert self.stage._cosign_tags == [] + + async def test_caches_probed_digests(self): + digest = "sha256:" + "a" * 64 + self.stage._tag_exists = AsyncMock(return_value=False) + + await self.stage._ensure_cosign_tags([digest]) + call_count = self.stage._tag_exists.await_count + + await self.stage._ensure_cosign_tags([digest]) + assert self.stage._tag_exists.await_count == call_count + + async def test_noop_when_bypass_taglist_false(self): + self.stage._bypass_taglist = False + self.stage._tag_exists = AsyncMock(return_value=True) + await self.stage._ensure_cosign_tags(["sha256:" + "a" * 64]) + assert self.stage._cosign_tags == [] + self.stage._tag_exists.assert_not_called() + + async def test_excludes_filter_candidates(self): + digest = "sha256:" + "a" * 64 + self.stage.remote.excludes = ["sha256-*"] + + self.stage._tag_exists = AsyncMock(return_value=True) + await self.stage._ensure_cosign_tags([digest]) + + assert self.stage._cosign_tags == [] + self.stage._tag_exists.assert_not_called() + + async def test_partial_excludes_filter_sig_only(self): + digest = "sha256:" + "a" * 64 + hex_part = "a" * 64 + self.stage.remote.excludes = ["*.sig"] + + self.stage._tag_exists = AsyncMock(return_value=True) + await self.stage._ensure_cosign_tags([digest]) + + found = set(self.stage._cosign_tags) + assert f"sha256-{hex_part}.sig" not in found + assert f"sha256-{hex_part}.att" in found + assert f"sha256-{hex_part}.sbom" in found + assert f"sha256-{hex_part}" in found + + async def test_child_manifest_digests_probed(self): + parent = "sha256:" + "a" * 64 + child = "sha256:" + "b" * 64 + + self.stage._tag_exists = AsyncMock(return_value=True) + await self.stage._ensure_cosign_tags([parent, child]) + + tags = set(self.stage._cosign_tags) + for hex_part in ["a" * 64, "b" * 64]: + for suffix in [".sig", ".att", ".sbom", ""]: + assert f"sha256-{hex_part}{suffix}" in tags + + async def test_caching_across_parent_and_child(self): + parent = "sha256:" + "a" * 64 + child = "sha256:" + "b" * 64 + + self.stage._tag_exists = AsyncMock(return_value=False) + + await self.stage._ensure_cosign_tags([parent]) + first_call_count = self.stage._tag_exists.await_count + + await self.stage._ensure_cosign_tags([parent, child]) + second_call_count = self.stage._tag_exists.await_count - first_call_count + + assert second_call_count == 4 + + async def test_rate_limit_error_propagates(self): + digest = "sha256:" + "a" * 64 + exc = aiohttp.ClientResponseError( + request_info=MagicMock(), history=(), status=429, message="Too Many Requests" + ) + self.stage._tag_exists = AsyncMock(side_effect=exc) + + with self.assertRaises(aiohttp.ClientResponseError): + await self.stage._ensure_cosign_tags([digest]) + + +class TestTagExistsHeadProbe(unittest.IsolatedAsyncioTestCase): + """Test _tag_exists HEAD request behavior.""" + + def setUp(self): + remote = MagicMock() + remote.policy = MagicMock() + remote.namespaced_upstream_name = "library/test" + remote.url = "https://registry.example/" + self.stage = ContainerFirstStage(remote=remote, signed_only=False) + + async def test_returns_true_on_success(self): + mock_downloader = AsyncMock() + mock_downloader.run = AsyncMock(return_value=MagicMock()) + self.stage.remote.get_downloader = MagicMock(return_value=mock_downloader) + + result = await self.stage._tag_exists("sha256-aaa.sig") + assert result is True + + async def test_returns_false_on_404(self): + mock_downloader = AsyncMock() + exc = aiohttp.ClientResponseError( + request_info=MagicMock(), history=(), status=404, message="Not Found" + ) + mock_downloader.run = AsyncMock(side_effect=exc) + self.stage.remote.get_downloader = MagicMock(return_value=mock_downloader) + + result = await self.stage._tag_exists("sha256-nonexistent.sig") + assert result is False + + async def test_raises_on_non_404_error(self): + mock_downloader = AsyncMock() + exc = aiohttp.ClientResponseError( + request_info=MagicMock(), history=(), status=401, message="Unauthorized" + ) + mock_downloader.run = AsyncMock(side_effect=exc) + self.stage.remote.get_downloader = MagicMock(return_value=mock_downloader) + + with self.assertRaises(aiohttp.ClientResponseError): + await self.stage._tag_exists("sha256-aaa.sig") + + if __name__ == "__main__": unittest.main()