fix(oci): normalize pinned references for registry inspection

.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'],

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)
 

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()