Add support for docker-credential-stores through environment

Sources/tart/Credentials/DockerConfigCredentialsProvider.swift

@@ -1,16 +1,54 @@
 import Foundation
 
 class DockerConfigCredentialsProvider: CredentialsProvider {
+
   func retrieve(host: String) throws -> (String, String)? {
-    let dockerConfigURL = FileManager.default.homeDirectoryForCurrentUser.appendingPathComponent(".docker").appendingPathComponent("config.json")
-    if !FileManager.default.fileExists(atPath: dockerConfigURL.path) {
+    guard let configFileURL = try configFileURL else {
       return nil
     }
-    let config = try JSONDecoder().decode(DockerConfig.self, from: Data(contentsOf: dockerConfigURL))
 
-    if let credentialsFromAuth = config.auths?[host]?.decodeCredentials() {
-      return credentialsFromAuth
+    let config = try JSONDecoder().decode(DockerConfig.self, from: Data(contentsOf: configFileURL))
+    return try retrieveCredentials(for: host, from: config)
+  }
+
+  // MARK: - Private
+  private var configFileURLFromEnvironment: URL? {
+    get throws {
+      guard let configPathFromEnvironment = ProcessInfo.processInfo.environment["TART_DOCKER_CONFIG"] else {
+        return nil
+      }
+
+      let url = URL(filePath: configPathFromEnvironment)
+
+      guard FileManager.default.fileExists(atPath: configPathFromEnvironment) else {
+        throw NSError.fileNotFoundError(url: url, message: "Registry authentication failed. Could not find docker configuration at '\(configPathFromEnvironment)'.")
+      }
+
+      return url
     }
+  }
+
+  private var dockerConfigFileURL: URL? {
+    let url = FileManager.default.homeDirectoryForCurrentUser.appendingPathComponent(".docker").appendingPathComponent("config.json")
+
+    guard FileManager.default.fileExists(atPath: url.path) else {
+      return nil
+    }
+
+    return url
+  }
+
+  private var configFileURL: URL? {
+    get throws {
+      return try configFileURLFromEnvironment ?? dockerConfigFileURL
+    }
+  }
+
+  private func retrieveCredentials(for host: String, from config: DockerConfig) throws -> (String, String)? {
+    if let credentials = config.auths?[host]?.decodeCredentials() {
+      return credentials
+    }
+
     if let helperProgram = try config.findCredHelper(host: host) {
       return try executeHelper(binaryName: "docker-credential-\(helperProgram)", host: host)
     }

Sources/tart/Utils.swift

@@ -6,6 +6,16 @@ extension Collection {
   }
 }
 
+extension NSError {
+  static func fileNotFoundError(url: URL, message: String = "") -> NSError {
+    return NSError(
+      domain: NSCocoaErrorDomain,
+      code: NSFileReadNoSuchFileError,
+      userInfo: [NSURLErrorKey: url, NSFilePathErrorKey: url.path, NSLocalizedFailureErrorKey: message]
+    )
+  }
+}
+
 func resolveBinaryPath(_ name: String) -> URL? {
   guard let path = ProcessInfo.processInfo.environment["PATH"] else {
     return nil

docs/integrations/vm-management.md

@@ -112,11 +112,17 @@ tart login acme.io
 If you login to your registry with OAuth, you may need to create an access token to use as the password.
 Credentials are securely stored in Keychain.
 
-In addition, Tart supports [Docker credential helpers](https://docs.docker.com/engine/reference/commandline/login/#credential-helpers)
-if defined in `~/.docker/config.json`.
+In addition, Tart supports Docker credential helpers via the `TART_DOCKER_CONFIG` environment variable or as a file 
+in `~/.docker/config.json`. If `TART_DOCKER_CONFIG` is set, Tart will attempt to load the credential configuration 
+from the specified file path. If the file cannot be found or the path is invalid, an error is thrown. If the variable 
+is not set, Tart will default to using `~/.docker/config.json`.
 
-Finally, `TART_REGISTRY_USERNAME` and `TART_REGISTRY_PASSWORD` environment variables allow to override authorization
-for all registries which might useful for integrating with your CI's secret management.
+Using `TART_DOCKER_CONFIG` provides greater flexibility, allowing multiple Tart instances to run with different 
+credential helper configurations—useful in CI/CD environments.
+
+Finally, `TART_REGISTRY_USERNAME` and `TART_REGISTRY_PASSWORD` environment variables allow to override any authorization
+for all registries which might useful for integrating with your CI's secret management. No additional lookup for a host-specific authorization
+with docker-credential-helpers is performed if these environment variables are set.
 
 ### Pushing a Local Image
 

integration-tests/conftest.py

@@ -14,3 +14,18 @@ def tart():
 def docker_registry():
     with DockerRegistry() as docker_registry:
         yield docker_registry
+
+@pytest.fixture(scope="function")
+def docker_registry_authenticated(request):
+    """
+    Provides an authenticated Docker registry where username/password
+    can be passed dynamically via the test case.
+
+    Usage:
+      - Add `docker_registry_authenticated` as a test argument.
+      - Pass `request.param = (username, password)` from the test.
+    """
+    credentials = request.param if hasattr(request, "param") else ("testuser", "testpassword")
+
+    with DockerRegistry(credentials=credentials) as docker_registry:
+        yield docker_registry

integration-tests/docker_registry.py

@@ -1,4 +1,7 @@
 import requests
+import tempfile
+import subprocess
+import bcrypt
 
 from testcontainers.core.waiting_utils import wait_container_is_ready
 from testcontainers.core.container import DockerContainer
@@ -7,9 +10,38 @@
 class DockerRegistry(DockerContainer):
     _default_exposed_port = 5000
 
-    def __init__(self):
+    def __init__(self, credentials: tuple[str, str] = None):
+        """
+        Initializes the DockerRegistry container.
+
+        :param credentials: A tuple (username, password). If None, starts the registry without authentication.
+        """
         super().__init__("registry:2")
         self.with_exposed_ports(self._default_exposed_port)
+        self.credentials = credentials
+
+        if credentials:
+            self._configure_basic_auth(credentials)
+
+    def _configure_basic_auth(self, credentials: tuple[str, str]):
+        username, password = credentials
+
+        # Set required environment variables for basic auth
+        self.with_env("REGISTRY_AUTH", "htpasswd")
+        self.with_env("REGISTRY_AUTH_HTPASSWD_PATH", "/auth/htpasswd")
+        self.with_env("REGISTRY_AUTH_HTPASSWD_REALM", "Registry Realm")
+
+        # Generate and mount the htpasswd file
+        htpasswd_path = self._generate_htpasswd(username, password)
+        self.with_volume_mapping(htpasswd_path, "/auth/htpasswd")
+
+    def _generate_htpasswd(self, username: str, password: str) -> str:
+        temp_file = tempfile.NamedTemporaryFile(delete=False)
+        hashed = bcrypt.hashpw(password.encode('utf-8'), bcrypt.gensalt()).decode('utf-8')
+        htpasswd_entry = f"{username}:{hashed}"
+        temp_file.write(htpasswd_entry.encode('utf-8'))
+        temp_file.close()
+        return temp_file.name
 
     @wait_container_is_ready(requests.exceptions.ConnectionError)
     def remote_name(self, for_vm: str):
@@ -18,3 +50,11 @@ def remote_name(self, for_vm: str):
         requests.get(f"http://127.0.0.1:{exposed_port}/v2/")
 
         return f"127.0.0.1:{exposed_port}/tart/{for_vm}:latest"
+
+    @wait_container_is_ready(requests.exceptions.ConnectionError)
+    def remote_host(self):
+        exposed_port = self.get_exposed_port(self._default_exposed_port)
+
+        requests.get(f"http://127.0.0.1:{exposed_port}/v2/")
+
+        return f"127.0.0.1:{exposed_port}"

integration-tests/requirements.txt

@@ -4,3 +4,4 @@ requests == 2.31.0 # work around https://github.com/psf/requests/issues/6707
 bitmath
 pytest-dependency
 paramiko
+bcrypt

integration-tests/tart.py

@@ -21,15 +21,17 @@ def __exit__(self, exc_type, exc_val, exc_tb):
     def home(self) -> str:
         return self.tmp_dir.name
 
-    def run(self, args):
-        env = os.environ.copy()
-        env.update({"TART_HOME": self.tmp_dir.name})
+    def run(self, args, env = {}, raise_on_nonzero_returncode=True):
+        environ = os.environ.copy()
+        environ.update(env)
+        environ.update({"TART_HOME": self.tmp_dir.name})
 
-        completed_process = subprocess.run(["tart"] + args, env=env, capture_output=True)
+        completed_process = subprocess.run(["tart"] + args, env=environ, capture_output=True)
 
-        completed_process.check_returncode()
+        if raise_on_nonzero_returncode:
+            completed_process.check_returncode
 
-        return completed_process.stdout.decode("utf-8"), completed_process.stderr.decode("utf-8")
+        return completed_process.stdout.decode("utf-8"), completed_process.stderr.decode("utf-8"), completed_process.returncode
 
     def run_async(self, args) -> subprocess.Popen:
         env = os.environ.copy()

integration-tests/test_clone.py

@@ -6,5 +6,5 @@ def test_clone(tart):
     tart.run(["clone", "debian", "ubuntu"])
 
     # Ensure that we have now 2 VMs
-    stdout, _, = tart.run(["list", "--source", "local", "--quiet"])
+    stdout, _, _ = tart.run(["list", "--source", "local", "--quiet"])
     assert stdout == "debian\nubuntu\n"

integration-tests/test_create.py

@@ -3,7 +3,7 @@ def test_create_macos(tart):
     tart.run(["create", "--from-ipsw", "latest", "macos-vm"])
 
     # Ensure that the VM was created
-    stdout, _ = tart.run(["list", "--source", "local", "--quiet"])
+    stdout, _, _ = tart.run(["list", "--source", "local", "--quiet"])
     assert stdout == "macos-vm\n"
 
 
@@ -12,5 +12,5 @@ def test_create_linux(tart):
     tart.run(["create", "--linux", "linux-vm"])
 
     # Ensure that the VM was created
-    stdout, _ = tart.run(["list", "--source", "local", "--quiet"])
+    stdout, _, _ = tart.run(["list", "--source", "local", "--quiet"])
     assert stdout == "linux-vm\n"

integration-tests/test_delete.py

@@ -3,12 +3,12 @@ def test_delete(tart):
     tart.run(["create", "--linux", "debian"])
 
     # Ensure that the VM exists
-    stdout, _, = tart.run(["list", "--source", "local", "--quiet"])
+    stdout, _, _ = tart.run(["list", "--source", "local", "--quiet"])
     assert stdout == "debian\n"
 
     # Delete the VM
     tart.run(["delete", "debian"])
 
     # Ensure that the VM was removed
-    stdout, _, = tart.run(["list", "--source", "local", "--quiet"])
+    stdout, _, _ = tart.run(["list", "--source", "local", "--quiet"])
     assert stdout == ""

integration-tests/test_oci.py

@@ -1,3 +1,5 @@
+import base64
+import json
 import os
 import tempfile
 import timeit
@@ -28,6 +30,72 @@ def test_pull_speed(self, tart, vm_with_random_disk, docker_registry):
 
         actual_speed_per_second = self._calculate_speed_per_second(amount_to_transfer, stop - start)
         assert actual_speed_per_second > minimal_speed_per_second
+
+    @pytest.mark.dependency()
+    @pytest.mark.parametrize("docker_registry_authenticated", [("user1", "pass1")], indirect=True)
+    def test_authenticated_push_from_env_config(self, tart, vm_with_random_disk, docker_registry_authenticated):
+        with tempfile.NamedTemporaryFile(delete=False) as tf:
+            tf.write(_docker_credentials_store(docker_registry_authenticated.remote_host(), "user1", "pass1"))
+            tf.close()
+        tart.run(["push", "--insecure", vm_with_random_disk, docker_registry_authenticated.remote_name(vm_with_random_disk)], env = { "TART_DOCKER_CONFIG": tf.name })
+
+    @pytest.mark.dependency()
+    @pytest.mark.parametrize("docker_registry_authenticated", [("user1", "pass1")], indirect=True)
+    def test_authenticated_push_from_docker_config(self, tart, vm_with_random_disk, docker_registry_authenticated):
+        with tempfile.NamedTemporaryFile(delete=False) as tf:
+            tf.write(_docker_credentials_store(docker_registry_authenticated.remote_host(), "user1", "pass1"))
+            tf.close()
+            if not os.path.exists(os.path.expanduser("~/.docker")):
+                os.mkdir(os.path.expanduser("~/.docker"))
+            os.rename(tf.name, os.path.expanduser("~/.docker/config.json"))
+
+        tart.run(["push", "--insecure", vm_with_random_disk, docker_registry_authenticated.remote_name(vm_with_random_disk)])
+
+    @pytest.mark.dependency()
+    @pytest.mark.parametrize("docker_registry_authenticated", [("user1", "pass1")], indirect=True)
+    def test_authenticated_push_env_path_precedence(self, tart, vm_with_random_disk, docker_registry_authenticated):
+        with tempfile.NamedTemporaryFile(delete=False) as tf:
+            tf.write(_docker_credentials_store(docker_registry_authenticated.remote_host(), "user1", "pass1"))
+            tf.close()
+
+        with tempfile.NamedTemporaryFile(delete=False) as tf2:
+            tf2.write(_docker_credentials_store(docker_registry_authenticated.remote_host(), "notuser", "notpassword"))
+            tf2.close()
+            if not os.path.exists(os.path.expanduser("~/.docker")):
+                os.mkdir(os.path.expanduser("~/.docker"))
+            os.rename(tf2.name, os.path.expanduser("~/.docker/config.json"))
+
+        tart.run(["push", "--insecure", vm_with_random_disk, docker_registry_authenticated.remote_name(vm_with_random_disk)], env = { "TART_DOCKER_CONFIG": tf.name })
+
+    @pytest.mark.dependency()
+    @pytest.mark.parametrize("docker_registry_authenticated", [("user1", "pass1")], indirect=True)
+    def test_authenticated_push_env_credentials_precedence(self, tart, vm_with_random_disk, docker_registry_authenticated):
+        with tempfile.NamedTemporaryFile(delete=False) as tf:
+            tf.write(_docker_credentials_store(docker_registry_authenticated.remote_host(), "notuser", "notpassword"))
+            tf.close()
+
+        env = {
+            "TART_REGISTRY_USERNAME": "user1",
+            "TART_REGISTRY_PASSWORD": "pass1",
+            "TART_DOCKER_CONFIG": tf.name
+        }
+        tart.run(["push", "--insecure", vm_with_random_disk, docker_registry_authenticated.remote_name(vm_with_random_disk)], env)
+
+    @pytest.mark.dependency()
+    @pytest.mark.parametrize("docker_registry_authenticated", [("user1", "pass1")], indirect=True)
+    def test_authenticated_push_invalid_env_path_error(self, tart, vm_with_random_disk, docker_registry_authenticated):
+        env = { "TART_DOCKER_CONFIG": "/temp/this-file-does-not-exist" }
+
+        _, stderr, returncode = tart.run(
+            ["push", "--insecure", vm_with_random_disk, docker_registry_authenticated.remote_name(vm_with_random_disk)],
+            env,
+            raise_on_nonzero_returncode=False
+        )
+
+        expected_error = f"Registry authentication failed. Could not find docker configuration at '/temp/this-file-does-not-exist'."
+
+        assert returncode == 1, f"Tart should fail with exit code 1 but failed with {returncode}."
+        assert expected_error in stderr, f"Expected error '{expected_error}' not found in stderr: {stderr}"
 
     @staticmethod
     def _calculate_speed_per_second(amount_transferred, time_taken):
@@ -53,3 +121,20 @@ def vm_with_random_disk(tart):
     yield vm_name
 
     tart.run(["delete", vm_name])
+
+def _docker_credentials_store(host, user, password):
+    # Encode "username:password" in Base64
+    auth_string = f"{user}:{password}"
+    auth_b64 = base64.b64encode(auth_string.encode()).decode()
+
+    # Create JSON structure
+    docker_auth = {
+        "auths": {
+            host: {
+                "auth": auth_b64
+            }
+        }
+    }
+
+    # Convert dictionary to JSON
+    return json.dumps(docker_auth).encode()

integration-tests/test_rename.py

@@ -6,5 +6,5 @@ def test_rename(tart):
     tart.run(["rename", "debian", "ubuntu"])
 
     # Ensure that the VM is now named "ubuntu"
-    stdout, _, = tart.run(["list", "--source", "local", "--quiet"])
+    stdout, _, _ = tart.run(["list", "--source", "local", "--quiet"])
     assert stdout == "ubuntu\n"

integration-tests/test_run.py

@@ -15,7 +15,7 @@ def test_run(tart, run_opts):
     tart_run_process = tart.run_async(["run", vm_name] + run_opts)
 
     # Obtain the VM's IP
-    stdout, _ = tart.run(["ip", vm_name, "--wait", "120"])
+    stdout, _, _ = tart.run(["ip", vm_name, "--wait", "120"])
     ip = stdout.strip()
 
     # Connect to the VM over SSH and shutdown it
@@ -29,4 +29,4 @@ def test_run(tart, run_opts):
     assert tart_run_process.returncode == 0
 
     # Delete the VM
-    _, _ = tart.run(["delete", vm_name])
+    _, _, _ = tart.run(["delete", vm_name])