From 3d1c542a056d07be7470b07c9d6e04fadbcb8dc5 Mon Sep 17 00:00:00 2001 From: Jonatan Date: Wed, 28 Oct 2026 10:10:57 +0100 Subject: [PATCH] fix(oci): normalize pinned references for registry inspection --- .github/scripts/tests/test_oci_wording.py | 3 +- oci/src/proxmenux_oci/custom.py | 24 +++++-- oci/tests/test_custom_registry_transport.py | 79 +++++++++++++++++++++ 3 files changed, 99 insertions(+), 7 deletions(-) create mode 100644 oci/tests/test_custom_registry_transport.py diff --git a/.github/scripts/tests/test_oci_wording.py b/.github/scripts/tests/test_oci_wording.py index 5d108f9c..67f6dda9 100644 --- a/.github/scripts/tests/test_oci_wording.py +++ b/.github/scripts/tests/test_oci_wording.py @@ -111,7 +111,8 @@ class OciWordingTests(TestCase): def test_registry_nonzero_and_success_command_unchanged(self): result = SimpleNamespace(returncode=1, stdout='', stderr='connection timed out') run = Mock(return_value=result) - report = function(CUSTOM, 'registry_report', subprocess=SimpleNamespace(run=run)) + report = function(CUSTOM, 'registry_report', subprocess=SimpleNamespace(run=run), + _registry_lookup=lambda reference: reference) self.assertEqual(report('example:latest'), (False, 'Could not inspect the image in its registry: example:latest')) run.assert_called_once_with(['skopeo', 'inspect', '--raw', 'docker://example:latest'], diff --git a/oci/src/proxmenux_oci/custom.py b/oci/src/proxmenux_oci/custom.py index b3b5a3f2..39347d5b 100644 --- a/oci/src/proxmenux_oci/custom.py +++ b/oci/src/proxmenux_oci/custom.py @@ -34,6 +34,18 @@ def _valid_image_reference(reference: str) -> bool: return False +def _registry_lookup(reference: str) -> str: + """Use the pull engine's repository/digest form only for registry transport. + + Skopeo rejects Docker's valid tag@digest spelling. The written reference + remains the recipe identity; its tag is removed only from the query. + """ + if not _valid_image_reference(reference): + raise ConversionError(f"{translate('This is not a valid image reference:')} {reference}") + image = _image_contract(reference) + return image['repository'] + '@' + image['digest'] if image.get('digest') else reference + + MAX_COMPOSE_BYTES = 256 * 1024 # An oci-compose: the file ProxMenux installs an application from. The catalog # keeps one for each of its applications and every installation records its own. @@ -254,7 +266,7 @@ def enrich_image_defaults(ui, document: dict[str, Any], *, image_entrypoints: di if not _valid_image_reference(reference): raise ConversionError(f"{translate('This is not a valid image reference:')} {reference}") if reference not in inspected: - command = ['skopeo', 'inspect', '--config', f'docker://{reference}'] + command = ['skopeo', 'inspect', '--config', f'docker://{_registry_lookup(reference)}'] result = subprocess.run(command, capture_output=True, text=True, check=False, timeout=120) if result.returncode and _registry_refused(reference): from .registry_accounts import offer_sign_in @@ -297,7 +309,7 @@ def compose_from_image(reference: str, title: str | None = None) -> str: persistent paths and its variables.""" if not _valid_image_reference(reference): raise ConversionError(f"{translate('This is not a valid image reference:')} {reference}") - result = subprocess.run(['skopeo', 'inspect', '--config', f'docker://{reference}'], + result = subprocess.run(['skopeo', 'inspect', '--config', f'docker://{_registry_lookup(reference)}'], capture_output=True, text=True, check=False, timeout=120) if result.returncode != 0: raise ConversionError(f"{translate('The image could not be read from its registry:')} " @@ -711,7 +723,7 @@ def ignored_settings(text: str) -> list[str]: def registry_report(reference: str) -> tuple[bool, str]: """Whether the image really exists in its registry, and for which architectures. A name written by hand is easy to get wrong.""" - result = subprocess.run(['skopeo', 'inspect', '--raw', f'docker://{reference}'], + result = subprocess.run(['skopeo', 'inspect', '--raw', f'docker://{_registry_lookup(reference)}'], capture_output=True, text=True, check=False, timeout=120) if result.returncode != 0: return False, f"{translate('Could not inspect the image in its registry:')} {reference}" @@ -722,7 +734,7 @@ def registry_report(reference: str) -> tuple[bool, str]: manifests = document.get('manifests') if not manifests: # The image publishes a single manifest: its architecture is in the image itself. - detail = subprocess.run(['skopeo', 'inspect', f'docker://{reference}'], + detail = subprocess.run(['skopeo', 'inspect', f'docker://{_registry_lookup(reference)}'], capture_output=True, text=True, check=False, timeout=120) try: architecture = json.loads(detail.stdout).get('Architecture') if detail.returncode == 0 else None @@ -740,9 +752,9 @@ def registry_report(reference: str) -> tuple[bool, str]: def _registry_refused(reference: str) -> bool: """Whether the registry answered that the image needs an account.""" try: - result = subprocess.run(['skopeo', 'inspect', '--raw', f'docker://{reference}'], + result = subprocess.run(['skopeo', 'inspect', '--raw', f'docker://{_registry_lookup(reference)}'], capture_output=True, text=True, check=False, timeout=120) - except (OSError, subprocess.TimeoutExpired): + except (OSError, subprocess.TimeoutExpired, ConversionError): return False return result.returncode != 0 and refuses_access(result.stderr) diff --git a/oci/tests/test_custom_registry_transport.py b/oci/tests/test_custom_registry_transport.py new file mode 100644 index 00000000..2f463a10 --- /dev/null +++ b/oci/tests/test_custom_registry_transport.py @@ -0,0 +1,79 @@ +"""Registry queries accept pinned tag@digest images without rewriting recipes.""" +from pathlib import Path +import json +import sys +from types import SimpleNamespace +import unittest +from unittest.mock import patch +import yaml + +ROOT = Path(__file__).resolve().parents[1] +sys.path.insert(0, str(ROOT / 'src')) +from proxmenux_oci import custom +from proxmenux_oci.converter import ConversionError + +DIGEST = 'sha256:' + 'a' * 64 +REFERENCE = 'registry.test:5443/team/app:release@' + DIGEST +LOOKUP = 'registry.test:5443/team/app@' + DIGEST + +class CustomRegistryTransportTests(unittest.TestCase): + def test_existing_image_parser_keeps_ports_and_pinned_digest(self): + pairs = [(REFERENCE, LOOKUP), ('[::1]:5443/team/app:release@' + DIGEST, + '[::1]:5443/team/app@' + DIGEST), + ('alpine:3', 'alpine:3'), ('registry.test:5443/team/app', 'registry.test:5443/team/app'), + ('ghcr.io/team/app@' + DIGEST, 'ghcr.io/team/app@' + DIGEST)] + for written, expected in pairs: + with self.subTest(written=written): + self.assertEqual(custom._registry_lookup(written), expected) + + def test_registry_report_normalizes_raw_and_platform_detail_queries(self): + responses = [SimpleNamespace(returncode=0, stdout='{"schemaVersion":2}'), + SimpleNamespace(returncode=0, stdout='{"Architecture":"amd64"}')] + with patch.object(custom.subprocess, 'run', side_effect=responses) as run: + available, report = custom.registry_report(REFERENCE) + self.assertTrue(available) + self.assertIn('amd64', report) + self.assertEqual([call.args[0][-1] for call in run.call_args_list], ['docker://' + LOOKUP] * 2) + + def test_failed_registry_report_keeps_the_written_reference(self): + with patch.object(custom.subprocess, 'run', return_value=SimpleNamespace(returncode=1, stdout='')): + available, report = custom.registry_report(REFERENCE) + self.assertFalse(available) + self.assertIn(REFERENCE, report) + + def test_image_metadata_cache_and_entrypoint_identity_keep_written_reference(self): + document = {'services': {'worker': {'image': REFERENCE}, 'worker2': {'image': REFERENCE}}} + defaults = {'config': {'Entrypoint': ['image-entrypoint'], 'ExposedPorts': {'8080/tcp': {}}, + 'Healthcheck': {'Test': ['CMD', 'healthcheck']}}} + entrypoints = {} + with patch.object(custom.subprocess, 'run', return_value=SimpleNamespace( + returncode=0, stdout=json.dumps(defaults))) as run: + enriched = custom.enrich_image_defaults(None, document, image_entrypoints=entrypoints) + run.assert_called_once() + self.assertEqual(run.call_args.args[0][-1], 'docker://' + LOOKUP) + self.assertEqual(entrypoints, {REFERENCE: ['image-entrypoint']}) + self.assertTrue(all(service['image'] == REFERENCE for service in enriched['services'].values())) + self.assertEqual(enriched['services']['worker']['healthcheck']['test'], ['CMD', 'healthcheck']) + + def test_image_to_compose_queries_digest_but_preserves_written_image(self): + with patch.object(custom.subprocess, 'run', return_value=SimpleNamespace( + returncode=0, stdout='{"config":{}}')) as run: + document = yaml.safe_load(custom.compose_from_image(REFERENCE)) + self.assertEqual(run.call_args.args[0][-1], 'docker://' + LOOKUP) + self.assertEqual(next(iter(document['services'].values()))['image'], REFERENCE) + + def test_private_registry_refusal_uses_the_same_transport(self): + with patch.object(custom.subprocess, 'run', return_value=SimpleNamespace( + returncode=1, stderr='unauthorized: authentication required')) as run: + self.assertTrue(custom._registry_refused(REFERENCE)) + self.assertEqual(run.call_args.args[0][-1], 'docker://' + LOOKUP) + + def test_invalid_references_are_rejected_before_any_registry_query(self): + for reference in ('ghcr.io/app@sha256:short', 'user:token@ghcr.io/app', 'ghcr.io/app;id'): + with self.subTest(reference=reference), patch.object(custom.subprocess, 'run') as run: + with self.assertRaises(ConversionError): + custom._registry_lookup(reference) + with self.assertRaises(ConversionError): + custom.registry_report(reference) + self.assertFalse(custom._registry_refused(reference)) + run.assert_not_called()