feat: use temp file for oci auth

This adjusts handling of oci auth passed via environment variable so
that we don't subsequently pass the value as a subprocess argument, as
this poses a security risk. Instead we write it to a temporary file,
which is automatically deleted after use.

Signed-off-by: Jon Burdo <[email protected]>

Author: jonburdo

clients/python/poetry.lock

@@ -48,6 +48,7 @@ sphinx-autobuild = ">=2021.3.14,<2025.0.0"
 pytest = ">=7.4.2,<9.0.0"
 coverage = { extras = ["toml"], version = "^7.3.2" }
 pytest-cov = ">=4.1,<7.0"
+pytest-mock = ">=3.7.0"
 ruff = ">=0.5.2,<0.13.0"
 mypy = "^1.7.0"
 # atm Ray is only available <3.13, so we will E2E test using Ray in compatible py environments.

clients/python/pyproject.toml

@@ -9,10 +9,11 @@
 import shutil
 import tempfile
 import threading
+from contextlib import AbstractContextManager, contextmanager
 from dataclasses import asdict, dataclass
 from pathlib import Path
 from subprocess import CalledProcessError
-from typing import TYPE_CHECKING, Callable, Protocol, TypeVar
+from typing import TYPE_CHECKING, Callable, Protocol, TextIO, TypeVar
 
 from typing_extensions import Literal, overload
 
@@ -223,15 +224,17 @@ def _backend_specific_params(
     # Determine backend
     if backend == "skopeo":
         prefix = "--src" if type == "pull" else "--dest"
+        auth_suffix = "authfile"
     elif backend == "oras":
         prefix = "--from" if type == "pull" else "--to"
+        auth_suffix = "registry-config"
+    else:
+        msg = f"invalid backend: {backend!r}"
+        raise ValueError(msg)
 
     # Actual param specifications
-    if username := kwargs.pop("username", None):
-        kwargs[f"{prefix}-username"] = username
-
-    if password := kwargs.pop("password", None):
-        kwargs[f"{prefix}-password"] = password
+    if authfile := kwargs.pop("authfile", None):
+        kwargs[f"{prefix}-{auth_suffix}"] = authfile
 
     return kwargs
 
@@ -344,16 +347,14 @@ def save_to_oci_registry(  # noqa: C901 ( complex args >8 )
         raise StoreError(msg) from e
 
     # Check for OCI Auth Env and a default
-    auth: str = None
+    auth: str | None = None
     if oci_auth_env_var:
-        env_value = _validate_env_var(oci_auth_env_var)
-        auth = _extract_auth_json(env_value)
-    else:
-        try:
-            env_value = _validate_env_var(".dockerconfigjson")
-            auth = _extract_auth_json(env_value)
-        except ValueError:
-            pass
+        auth = _validate_env_var(oci_auth_env_var)
+    elif ".dockerconfigjson" in os.environ:
+        auth = os.environ[".dockerconfigjson"] # noqa: SIM112
+
+    elif oci_username and oci_password:
+        auth = json.dumps(create_auth_object(oci_ref, oci_username, oci_password))
 
     # If a custom backend is provided, use it, else fetch the backend out of the registry
     if custom_oci_backend:
@@ -374,30 +375,48 @@ def save_to_oci_registry(  # noqa: C901 ( complex args >8 )
         dest_dir_cleanup = True
     local_image_path = Path(dest_dir)
 
-    # Set params
     params = {}
-
-    # User/pass
-    if auth:
-        usr_pass = auth.split(":")
-        params["username"] = usr_pass[0]
-        params["password"] = usr_pass[-1]
-    elif oci_username and oci_password:
-        params["username"] = oci_username
-        params["password"] = oci_password
-
-    backend_def.pull(base_image, local_image_path, **params)
-    # Extract the absolute path from the files found in the path
-    files = [file[0] for file in _get_files_from_path(model_files_path)]
-    oci_layers_on_top(local_image_path, files, modelcard)
-    backend_def.push(local_image_path, oci_ref, **params)
+    with temp_auth_file(auth) as auth_file:
+        if auth_file is not None:
+            params["authfile"] = auth_file.name
+        backend_def.pull(base_image, local_image_path, **params)
+        # Extract the absolute path from the files found in the path
+        files = [file[0] for file in _get_files_from_path(model_files_path)]
+        oci_layers_on_top(local_image_path, files, modelcard)
+        backend_def.push(local_image_path, oci_ref, **params)
 
     # Return the OCI URI
     if dest_dir_cleanup:
         shutil.rmtree(dest_dir)
     return f"oci://{oci_ref}"
 
 
+@overload
+def temp_auth_file(auth: str) -> AbstractContextManager[TextIO]:
+    ...
+
+
+@overload
+def temp_auth_file(auth: None) -> AbstractContextManager[None]:
+    ...
+
+
+@contextmanager
+def temp_auth_file(auth: str | None) -> AbstractContextManager[TextIO | None]:
+    """Create a temporary auth file with optional auth data.
+
+    If auth is None, yields None. Otherwise creates a temporary JSON file
+    containing the auth string and yields the file handle.
+    """
+    if auth is None:
+        yield None
+    else:
+        with tempfile.NamedTemporaryFile(mode="w+", encoding="utf-8", suffix=".json") as temp_auth_file:
+            temp_auth_file.write(auth)
+            temp_auth_file.flush()
+            yield temp_auth_file
+
+
 def _s3_creds(
     endpoint_url: str | None = None,
     access_key_id: str | None = None,
@@ -636,6 +655,66 @@ def _extract_auth_json(auth_data: str) -> str:
         raise ValueError(invalid_json_msg) from e
 
 
+def get_auth_reference(image_path: str) -> str:
+    """Parses an arbitrary container image path to extract a valid reference.
+
+    for use as a key in a container registry auth.json file.
+
+    Examples:
+        'quay.io/my-org/my-registry:1.0.0' -> 'quay.io/my-org/my-registry'
+        'my-private-registry:5000/team/app:latest' -> 'my-private-registry:5000/team/app'
+        'ubuntu' -> 'docker.io'
+        'ubuntu:22.04' -> 'docker.io'
+        'my-user/my-app' -> 'docker.io'
+        'my-user/my-app:v2' -> 'docker.io'
+        'quay.io/my-org/my-registry@sha256:f1b3f5a2d...' -> 'quay.io/my-org/my-registry'
+        'localhost:5000/my-local-image' -> 'localhost:5000/my-local-image'
+        'localhost:5000/my-local-image:test-tag' -> 'localhost:5000/my-local-image'
+    """
+    repo_path = image_path
+
+    # Remove digest if it exists
+    if "@" in repo_path:
+        repo_path = repo_path.split("@", 1)[0]
+
+    # Separate the tag from the repository path.
+    # The tag is what comes after the last colon, but only if that colon
+    # is not part of the hostname/port. A colon indicates a tag if it
+    # appears after the last slash in the path.
+    last_colon = repo_path.rfind(":")
+    last_slash = repo_path.rfind("/")
+
+    if last_colon > last_slash:
+        # This is a tag, not a port, so we strip it.
+        repo_path = repo_path[:last_colon]
+
+    # Handle default Docker Hub images (e.g., 'ubuntu', 'user/repo').
+    # The hostname is the part of the path before the first slash.
+    first_slash_index = repo_path.find("/")
+    hostname = repo_path
+    if first_slash_index != -1:
+        hostname = repo_path[:first_slash_index]
+
+    # If the hostname part doesn't contain a '.' (like quay.io) or a ':' (like localhost:5000),
+    # it's a short name for an image on Docker Hub.
+    if "." not in hostname and ":" not in hostname:
+        return "docker.io"
+
+    # For all other images, the full repository path is the reference.
+    return repo_path
+
+
+def create_auth_object(oci_ref: str, username: str, password: str) -> dict[str: dict[str, dict[str, str]]]:
+    """Create an auth object for container registry authentication.
+
+    This object can be encoded as json with json.dumps() producing the
+    contents for valid authfile.
+    """
+    auth_ref = get_auth_reference(oci_ref)
+    auth_value = base64.b64encode(f"{username}:{password}".encode()).decode("utf-8")
+    return {"auths": {auth_ref: {"auth": auth_value}}}
+
+
 def rand_suffix(size: int = 8) -> str:
     """Generate a random suffix.
 

clients/python/src/model_registry/utils.py

@@ -1,4 +1,7 @@
+import json
 import os
+from contextlib import contextmanager
+from pathlib import Path
 
 import pytest
 
@@ -8,6 +11,7 @@
     _get_files_from_path,
     s3_uri_from,
     save_to_oci_registry,
+    temp_auth_file,
 )
 
 
@@ -124,6 +128,46 @@ def test_save_to_oci_registry_with_custom_backend(
     assert uri == f"oci://{oci_ref}"
 
 
+def test_save_to_oci_registry_with_username_password(mocker, tmp_path):
+    username = "user32"
+    password = "zi3327"
+    model_files_path = tmp_path / "model-files"
+    model_files_path.mkdir()
+    (model_files_path / "model.bin").touch()
+    dest_dir = tmp_path / "dest"
+
+    temp_auth_file_info = {}
+
+    @contextmanager
+    def temp_auth_file_wrapper(auth):
+        with temp_auth_file(auth) as f:
+            temp_auth_file_info["path"] = f.name
+            temp_auth_file_info["contents"] = Path(f.name).read_text()
+            yield f
+
+    mock_skopeo_pull = mocker.patch("olot.backend.skopeo.skopeo_pull")
+    mock_skopeo_push = mocker.patch("olot.backend.skopeo.skopeo_push")
+    mocker.patch("olot.basics.oci_layers_on_top")
+    mocker.patch("model_registry.utils.temp_auth_file", side_effect=temp_auth_file_wrapper)
+
+    save_to_oci_registry(
+        base_image="busybox",
+        oci_ref="quay.io/example/example:latest",
+        model_files_path=model_files_path,
+        dest_dir=dest_dir,
+        backend="skopeo",
+        oci_username=username,
+        oci_password=password,
+    )
+
+    assert mock_skopeo_pull.call_args.args == ("busybox", dest_dir, ["--src-authfile", mocker.ANY])
+    assert mock_skopeo_pull.call_args.kwargs == {}
+    assert mock_skopeo_push.call_args.args == (dest_dir, "quay.io/example/example:latest", ["--dest-authfile", mocker.ANY])
+    assert mock_skopeo_push.call_args.kwargs == {}
+    assert json.loads(temp_auth_file_info["contents"]) == {"auths": {"quay.io/example/example": {"auth": "dXNlcjMyOnppMzMyNw=="}}}
+    assert not Path(temp_auth_file_info["path"]).exists()
+
+
 @pytest.mark.e2e(type="oci")
 def test_save_to_oci_registry_auth_params(
     get_temp_dir_with_models,