fix: normalize URL passed to ContainerRegistry

Add methods for normalizing the URL passed to ContainerRegistry.

These changes ensure that special registry hostnames are replaced
even if the passed registry URL contains a scheme or a port number.

coregio/registry_api.py

@@ -1,7 +1,6 @@
 """Container registry API implementation."""
 import json
 import logging
-import re
 from typing import Any, Dict, List, Optional, Callable
 from urllib.parse import urljoin, urlparse
 
@@ -32,6 +31,7 @@
     "docker.io": "index.docker.io",
     "registry-1.docker.io": "index.docker.io",
     "hub.docker.com": "index.docker.io",
+    "registry.hub.docker.com": "index.docker.io",
 }
 
 
@@ -59,7 +59,7 @@ def __init__(
         docker_cfg (Optional, str): DockerConfigJson string
         """
 
-        self.url = SPECIAL_DOCKER_ALIASES.get(url, url)
+        self.url = self._normalize_registry_url(url)
         self._original_url = url
         self.docker_cfg = docker_cfg
 
@@ -68,6 +68,31 @@ def __init__(
 
         self.auth_header = None
 
+    @staticmethod
+    def _normalize_registry_url(url: str) -> str:
+        """
+        Normalize registry URL:
+        - If needed, the hostname is replaced according
+          to the SPECIAL_DOCKER_ALIASES mapping.
+        - Scheme is added if missing.
+        - Port number is preserved if present.
+        - Path is discarded if present.
+
+        Args:
+            url: Registry URL
+
+        Returns:
+            str: Normalized registry URL
+        """
+        url_with_scheme = utils.add_scheme_if_missing(url)
+        parsed = urlparse(url_with_scheme)
+        hostname = SPECIAL_DOCKER_ALIASES.get(parsed.hostname, parsed.hostname)
+
+        normalized_url = f"{parsed.scheme}://{hostname}"
+        if parsed.port:
+            normalized_url = f"{normalized_url}:{parsed.port}"
+        return normalized_url
+
     def _get_auth_token(self) -> Any:
         """
         Extract registry auth token from docker_config_json.
@@ -120,8 +145,7 @@ def _select_registry_auth_token_from_docker_config(
                 return auth
 
             # Add a schema to the URL if missing
-            if not re.search(r"^[A-Za-z0-9+.\-]+://", registry_key):
-                registry_key = f"https://{registry_key}"
+            registry_key = utils.add_scheme_if_missing(registry_key)
 
             # Parse URL and compare matching hostname
             parsed_key = urlparse(registry_key)
@@ -239,7 +263,7 @@ def get_request(
             Response: The resulting Response object
         """
 
-        full_url = urljoin(f"https://{self.url}", path)
+        full_url = urljoin(self.url, path)
 
         LOGGER.debug("Querying registry: GET %s %s %s", full_url, headers, params)
         resp = self._get(full_url, params=params, headers=headers)

coregio/utils.py

@@ -2,6 +2,7 @@
 Utilities for http queries
 """
 import logging
+import re
 from typing import Any
 
 from requests import Session
@@ -76,3 +77,18 @@ def add_session_retries(
     adapter = HTTPAdapter(max_retries=retries)
     session.mount("http://", adapter)
     session.mount("https://", adapter)
+
+
+def add_scheme_if_missing(url: str) -> str:
+    """
+    Add https:// to the url if it does not contain a scheme.
+
+    Args:
+        url: Url to check
+
+    Returns:
+        str: Url containing a scheme
+    """
+    if not re.search(r"^[A-Za-z0-9+.\-]+://", url):
+        return f"https://{url}"
+    return url

tests/test_registry_api.py

@@ -7,6 +7,33 @@
 from coregio.registry_api import ContainerRegistry
 
 
+@pytest.mark.parametrize(
+    ["url", "expected_url"],
+    [
+        ("quay.io", "https://quay.io"),
+        ("http://quay.io", "http://quay.io"),
+        ("quay.io:8080", "https://quay.io:8080"),
+        ("quay.io/path", "https://quay.io"),
+        ("http://quay.io:8080/path", "http://quay.io:8080"),
+        ("registry-1.docker.io", "https://index.docker.io"),
+        ("hub.docker.com", "https://index.docker.io"),
+        ("registry.hub.docker.com", "https://index.docker.io"),
+        ("docker.io", "https://index.docker.io"),
+        ("http://docker.io", "http://index.docker.io"),
+        ("docker.io:8080", "https://index.docker.io:8080"),
+        ("docker.io/path", "https://index.docker.io"),
+        ("http://docker.io:8080/path", "http://index.docker.io:8080"),
+    ],
+)
+def test__normalize_registry_url(
+    url: str,
+    expected_url: str,
+) -> None:
+    result_url = ContainerRegistry._normalize_registry_url(url)
+
+    assert result_url == expected_url
+
+
 @pytest.mark.parametrize(
     ["registry", "cfg", "auth", "token"],
     [

tests/test_utils.py

@@ -18,3 +18,19 @@ def test_add_session_retries() -> None:
 
     assert reps is None
     assert session.mount.call_count == 2
+
+
+@pytest.mark.parametrize(
+    ["url", "expected_url"],
+    [
+        ("quay.io", "https://quay.io"),
+        ("http://quay.io", "http://quay.io"),
+    ],
+)
+def test_add_scheme_if_missing(
+    url: str,
+    expected_url: str,
+) -> None:
+    result_url = utils.add_scheme_if_missing(url)
+
+    assert result_url == expected_url