fix(core): Improve private registry support (tolerate not implemented fields in DOCKER_AUTH_CONFIG) (#647)

Continuing #562, got some feedback regarding an issue with unsupported
use cases.
In this PR we will try to:
1. Map the use cases
2. Raise a warning regarding unsupported uses cases (hopefully they will
be added later)
3. Address/Fix the issue where unsupported JSON schema for
`DOCKER_AUTH_CONFIG` leads to an error

As always any feedback will be much appreciated.
_Please note this PR does not implement all use-cases just does a better
job at preparing and handling them for now_

Author: Tranquility2

core/testcontainers/core/auth.py

@@ -0,0 +1,88 @@
+import base64 as base64
+import json as json
+from collections import namedtuple
+from logging import warning
+from typing import Optional
+
+DockerAuthInfo = namedtuple("DockerAuthInfo", ["registry", "username", "password"])
+
+_AUTH_WARNINGS = {
+    "credHelpers": "DOCKER_AUTH_CONFIG is experimental, credHelpers not supported yet",
+    "credsStore": "DOCKER_AUTH_CONFIG is experimental, credsStore not supported yet",
+}
+
+
+def process_docker_auth_config_encoded(auth_config_dict: dict) -> list[DockerAuthInfo]:
+    """
+    Process the auths config.
+
+    Example:
+    {
+        "auths": {
+            "https://index.docker.io/v1/": {
+                "auth": "dXNlcm5hbWU6cGFzc3dvcmQ="
+            }
+        }
+    }
+
+    Returns a list of DockerAuthInfo objects.
+    """
+    auth_info: list[DockerAuthInfo] = []
+
+    auths = auth_config_dict.get("auths")
+    for registry, auth in auths.items():
+        auth_str = auth.get("auth")
+        auth_str = base64.b64decode(auth_str).decode("utf-8")
+        username, password = auth_str.split(":")
+        auth_info.append(DockerAuthInfo(registry, username, password))
+
+    return auth_info
+
+
+def process_docker_auth_config_cred_helpers(auth_config_dict: dict) -> None:
+    """
+    Process the credHelpers config.
+
+    Example:
+    {
+        "credHelpers": {
+            "<aws_account_id>.dkr.ecr.<region>.amazonaws.com": "ecr-login"
+        }
+    }
+
+    This is not supported yet.
+    """
+    if "credHelpers" in _AUTH_WARNINGS:
+        warning(_AUTH_WARNINGS.pop("credHelpers"))
+
+
+def process_docker_auth_config_store(auth_config_dict: dict) -> None:
+    """
+    Process the credsStore config.
+
+    Example:
+    {
+        "credsStore": "ecr-login"
+    }
+
+    This is not supported yet.
+    """
+    if "credsStore" in _AUTH_WARNINGS:
+        warning(_AUTH_WARNINGS.pop("credsStore"))
+
+
+def parse_docker_auth_config(auth_config: str) -> Optional[list[DockerAuthInfo]]:
+    """Parse the docker auth config from a string and handle the different formats."""
+    try:
+        auth_config_dict: dict = json.loads(auth_config)
+        if "credHelpers" in auth_config:
+            process_docker_auth_config_cred_helpers(auth_config_dict)
+        if "credsStore" in auth_config:
+            process_docker_auth_config_store(auth_config_dict)
+        if "auths" in auth_config:
+            return process_docker_auth_config_encoded(auth_config_dict)
+
+    except (json.JSONDecodeError, KeyError, ValueError) as exp:
+        raise ValueError("Could not parse docker auth config") from exp
+
+    return None

core/testcontainers/core/docker_client.py

@@ -24,9 +24,10 @@
 from docker.models.images import Image, ImageCollection
 from typing_extensions import ParamSpec
 
+from testcontainers.core.auth import DockerAuthInfo, parse_docker_auth_config
 from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.labels import SESSION_ID, create_labels
-from testcontainers.core.utils import default_gateway_ip, inside_container, parse_docker_auth_config, setup_logger
+from testcontainers.core.utils import default_gateway_ip, inside_container, setup_logger
 
 LOGGER = setup_logger(__name__)
 
@@ -67,8 +68,11 @@ def __init__(self, **kwargs) -> None:
         self.client.api.headers["x-tc-sid"] = SESSION_ID
         self.client.api.headers["User-Agent"] = "tc-python/" + importlib.metadata.version("testcontainers")
 
+        # Verify if we have a docker auth config and login if we do
         if docker_auth_config := get_docker_auth_config():
-            self.login(docker_auth_config)
+            LOGGER.debug(f"DOCKER_AUTH_CONFIG found: {docker_auth_config}")
+            if auth_config := parse_docker_auth_config(docker_auth_config):
+                self.login(auth_config[0])  # Only using the first auth config)
 
     @_wrapped_container_collection
     def run(
@@ -203,11 +207,10 @@ def host(self) -> str:
                 return ip_address
         return "localhost"
 
-    def login(self, docker_auth_config: str) -> None:
+    def login(self, auth_config: DockerAuthInfo) -> None:
         """
         Login to a docker registry using the given auth config.
         """
-        auth_config = parse_docker_auth_config(docker_auth_config)[0]  # Only using the first auth config
         login_info = self.client.login(**auth_config._asdict())
         LOGGER.debug(f"logged in using {login_info}")
 

core/testcontainers/core/utils.py

@@ -1,18 +1,13 @@
-import base64
-import json
 import logging
 import os
 import platform
 import subprocess
 import sys
-from collections import namedtuple
 
 LINUX = "linux"
 MAC = "mac"
 WIN = "win"
 
-DockerAuthInfo = namedtuple("DockerAuthInfo", ["registry", "username", "password"])
-
 
 def setup_logger(name: str) -> logging.Logger:
     logger = logging.getLogger(name)
@@ -82,29 +77,3 @@ def raise_for_deprecated_parameter(kwargs: dict, name: str, replacement: str) ->
     if kwargs.pop(name, None):
         raise ValueError(f"Use `{replacement}` instead of `{name}`")
     return kwargs
-
-
-def parse_docker_auth_config(auth_config: str) -> list[DockerAuthInfo]:
-    """
-    Parse the docker auth config from a string.
-
-    Example:
-    {
-        "auths": {
-            "https://index.docker.io/v1/": {
-                "auth": "dXNlcm5hbWU6cGFzc3dvcmQ="
-            }
-        }
-    }
-    """
-    auth_info: list[DockerAuthInfo] = []
-    try:
-        auth_config_dict: dict = json.loads(auth_config).get("auths")
-        for registry, auth in auth_config_dict.items():
-            auth_str = auth.get("auth")
-            auth_str = base64.b64decode(auth_str).decode("utf-8")
-            username, password = auth_str.split(":")
-            auth_info.append(DockerAuthInfo(registry, username, password))
-        return auth_info
-    except (json.JSONDecodeError, KeyError, ValueError) as exp:
-        raise ValueError("Could not parse docker auth config") from exp

core/tests/test_auth.py

@@ -0,0 +1,84 @@
+import json
+import pytest
+
+from testcontainers.core.auth import parse_docker_auth_config, DockerAuthInfo
+
+
+def test_parse_docker_auth_config_encoded():
+    auth_config_json = '{"auths":{"https://index.docker.io/v1/":{"auth":"dXNlcm5hbWU6cGFzc3dvcmQ="}}}'
+    auth_info = parse_docker_auth_config(auth_config_json)
+    assert len(auth_info) == 1
+    assert auth_info[0] == DockerAuthInfo(
+        registry="https://index.docker.io/v1/",
+        username="username",
+        password="password",
+    )
+
+
+def test_parse_docker_auth_config_cred_helpers():
+    auth_dict = {"credHelpers": {"<aws_account_id>.dkr.ecr.<region>.amazonaws.com": "ecr-login"}}
+    auth_config_json = json.dumps(auth_dict)
+    assert parse_docker_auth_config(auth_config_json) is None
+
+
+def test_parse_docker_auth_config_store():
+    auth_dict = {"credsStore": "ecr-login"}
+    auth_config_json = json.dumps(auth_dict)
+    assert parse_docker_auth_config(auth_config_json) is None
+
+
+def test_parse_docker_auth_config_encoded_multiple():
+    auth_dict = {
+        "auths": {
+            "localhost:5000": {"auth": "dXNlcjE6cGFzczE=="},
+            "https://example.com": {"auth": "dXNlcl9uZXc6cGFzc19uZXc=="},
+            "example2.com": {"auth": "YWJjOjEyMw==="},
+        }
+    }
+    auth_config_json = json.dumps(auth_dict)
+    auth_info = parse_docker_auth_config(auth_config_json)
+    assert len(auth_info) == 3
+    assert auth_info[0] == DockerAuthInfo(
+        registry="localhost:5000",
+        username="user1",
+        password="pass1",
+    )
+    assert auth_info[1] == DockerAuthInfo(
+        registry="https://example.com",
+        username="user_new",
+        password="pass_new",
+    )
+    assert auth_info[2] == DockerAuthInfo(
+        registry="example2.com",
+        username="abc",
+        password="123",
+    )
+
+
+def test_parse_docker_auth_config_unknown():
+    auth_config_str = '{"key": "value"}'
+    assert parse_docker_auth_config(auth_config_str) is None
+
+
+def test_parse_docker_auth_config_error():
+    auth_config_str = "bad//string"
+    with pytest.raises(ValueError):
+        parse_docker_auth_config(auth_config_str)
+
+
+def test_parse_docker_auth_all():
+    test_dict = {
+        "auths": {
+            "localhost:5000": {"auth": "dXNlcjE6cGFzczE=="},
+        },
+        "credHelpers": {"<aws_account_id>.dkr.ecr.<region>.amazonaws.com": "ecr-login"},
+        "credsStore": "ecr-login",
+    }
+    auth_config_json = json.dumps(test_dict)
+    assert parse_docker_auth_config(auth_config_json) == [
+        DockerAuthInfo(
+            registry="localhost:5000",
+            username="user1",
+            password="pass1",
+        )
+    ]

core/tests/test_docker_client.py

@@ -1,4 +1,5 @@
 import os
+import json
 from collections import namedtuple
 from unittest import mock
 from unittest.mock import MagicMock, patch
@@ -8,9 +9,11 @@
 from testcontainers.core.config import testcontainers_config as c
 from testcontainers.core.container import DockerContainer
 from testcontainers.core.docker_client import DockerClient
-from testcontainers.core.utils import parse_docker_auth_config
+from testcontainers.core.auth import parse_docker_auth_config
 from testcontainers.core.image import DockerImage
 
+from pytest import mark
+
 
 def test_docker_client_from_env():
     test_kwargs = {"test_kw": "test_value"}
@@ -48,6 +51,55 @@ def test_docker_client_login():
     mock_docker.from_env.return_value.login.assert_called_with(**{"value": "test"})
 
 
+def test_docker_client_login_empty_get_docker_auth_config():
+    mock_docker = MagicMock(spec=docker)
+    mock_get_docker_auth_config = MagicMock()
+    mock_get_docker_auth_config.return_value = None
+
+    with (
+        mock.patch.object(c, "_docker_auth_config", "test"),
+        patch("testcontainers.core.docker_client.docker", mock_docker),
+        patch("testcontainers.core.docker_client.get_docker_auth_config", mock_get_docker_auth_config),
+    ):
+        DockerClient()
+
+    mock_docker.from_env.return_value.login.assert_not_called()
+
+
+def test_docker_client_login_empty_parse_docker_auth_config():
+    mock_docker = MagicMock(spec=docker)
+    mock_parse_docker_auth_config = MagicMock(spec=parse_docker_auth_config)
+    mock_utils = MagicMock()
+    mock_utils.parse_docker_auth_config = mock_parse_docker_auth_config
+    mock_parse_docker_auth_config.return_value = None
+
+    with (
+        mock.patch.object(c, "_docker_auth_config", "test"),
+        patch("testcontainers.core.docker_client.docker", mock_docker),
+        patch("testcontainers.core.docker_client.parse_docker_auth_config", mock_parse_docker_auth_config),
+    ):
+        DockerClient()
+
+    mock_docker.from_env.return_value.login.assert_not_called()
+
+
+# This is used to make sure we don't fail (nor try to login) when we have unsupported auth config
+@mark.parametrize("auth_config_sample", [{"credHelpers": {"test": "login"}}, {"credsStore": "login"}])
+def test_docker_client_login_unsupported_auth_config(auth_config_sample):
+    mock_docker = MagicMock(spec=docker)
+    mock_get_docker_auth_config = MagicMock()
+    mock_get_docker_auth_config.return_value = json.dumps(auth_config_sample)
+
+    with (
+        mock.patch.object(c, "_docker_auth_config", "test"),
+        patch("testcontainers.core.docker_client.docker", mock_docker),
+        patch("testcontainers.core.docker_client.get_docker_auth_config", mock_get_docker_auth_config),
+    ):
+        DockerClient()
+
+    mock_docker.from_env.return_value.login.assert_not_called()
+
+
 def test_container_docker_client_kw():
     test_kwargs = {"test_kw": "test_value"}
     mock_docker = MagicMock(spec=docker)