feat(ENG-13678): adding cargo credential helper to cli

cloudsmith_cli/cli/commands/credential_helper/__init__.py

@@ -9,6 +9,7 @@
 import click
 
 from ..main import main
+from .cargo import cargo as cargo_cmd
 from .docker import docker as docker_cmd
 from .generic import generic as generic_cmd
 from .manage import install_cmd, list_cmd, uninstall_cmd
@@ -45,5 +46,6 @@ def credential_helper():
 credential_helper.add_command(install_cmd, name="install")
 credential_helper.add_command(uninstall_cmd, name="uninstall")
 credential_helper.add_command(list_cmd, name="list")
+credential_helper.add_command(cargo_cmd, name="cargo")
 
 main.add_command(credential_helper, name="credential-helper")

cloudsmith_cli/cli/commands/credential_helper/cargo.py

@@ -0,0 +1,68 @@
+# Copyright 2026 Cloudsmith Ltd
+"""
+Docker credential helper command.
+
+Implements the Docker credential helper protocol for Cloudsmith registries.
+
+See: https://github.com/docker/docker-credential-helpers
+"""
+
+import sys
+
+import click
+
+from ....credential_helpers.cargo import execute
+from ...decorators import common_api_auth_options, resolve_credentials
+
+
+@click.command()
+@click.argument("operation", required=False, default="get")
+@common_api_auth_options
+@resolve_credentials
+def cargo(opts, operation):
+    """
+    Cargo credential helper for Cloudsmith registries.
+
+    Reads a Docker registry server URL from stdin and returns credentials in
+    TOML format.  Implements the full Docker credential helper protocol
+    (get/store/erase/list).
+
+    Provides credentials for all Cloudsmith Docker registries: ``*.cloudsmith.io``,
+    ``*.cloudsmith.com``, and any custom domains configured for the organisation
+    (requires an organisation - ``--org``, CLOUDSMITH_ORG or ``org`` in
+    ``config.ini`` - and a valid API key/token).
+
+    Input (stdin):
+        Server URL as plain text (e.g. "docker.cloudsmith.io")
+
+    Output (stdout):
+        JSON: {"Username": "token", "Secret": "<cloudsmith-token>"}
+
+    Exit codes:
+        0: Success
+        1: Error (no credentials available, not a Cloudsmith registry, etc.)
+
+    Examples:
+        # Manual testing
+        $ echo "docker.cloudsmith.io" | cloudsmith credential-helper docker
+
+        # Called by Docker via launcher
+        $ echo "docker.cloudsmith.io" | docker-credential-cloudsmith get
+
+    Environment variables:
+        CLOUDSMITH_API_KEY: API key for authentication (optional)
+        CLOUDSMITH_ORG:     Organisation slug (required for custom domain support)
+    """
+    exit_code, stdout, stderr = execute(
+        operation,
+        sys.stdin,
+        credential=opts.credential,
+        api_host=opts.api_host,
+        org=opts.org,
+    )
+
+    if stdout is not None:
+        click.echo(stdout)
+    if stderr is not None:
+        click.echo(stderr, err=True)
+    sys.exit(exit_code)

cloudsmith_cli/cli/commands/credential_helper/manage.py

@@ -6,23 +6,24 @@
 ``config.json`` entries for each supported credential helper.
 """
 
 from __future__ import annotations
 
 import sys
 
 import click
 
 from cloudsmith_cli.credential_helpers.generic import PartialInstallError
 from cloudsmith_cli.credential_helpers.pnpm.installer import PNPMInstaller
+from cloudsmith_cli.credential_helpers.cargo.installer import CargoInstaller
 
 from ....credential_helpers.docker.installer import DockerInstaller
 from ... import utils
 from ...decorators import (
     common_api_auth_options,
     common_cli_config_options,
     common_cli_output_options,
     resolve_credentials,
 )
 
 # ---------------------------------------------------------------------------
 # Helper registry — extend here when new helpers are added
@@ -31,6 +32,7 @@
 _INSTALLERS: dict[str, type] = {
     "docker": DockerInstaller,
     "pnpm": PNPMInstaller,
+    "cargo": CargoInstaller,
 }
 
 

cloudsmith_cli/cli/tests/test_exceptions.py

@@ -8,7 +8,6 @@
 from cloudsmith_cli.core.api.exceptions import ApiException
 from cloudsmith_cli.core.credentials.models import CredentialResult
 
-
 API_KEY_HINT = (
     "This usually means your API key is invalid, expired, or lacks access to this "
     "resource - check your credentials and try again."

cloudsmith_cli/cli/tests/test_push.py

@@ -1490,21 +1490,23 @@ def test_wait_for_package_sync_json_mode_prints_status_reason_on_failure(capsys)
         status_reason,
     )
 
-    with patch(
-        "cloudsmith_cli.cli.commands.push.get_package_status",
-        return_value=failed_status,
+    with (
+        patch(
+            "cloudsmith_cli.cli.commands.push.get_package_status",
+            return_value=failed_status,
+        ),
+        pytest.raises(click.exceptions.Exit) as exc_info,
     ):
-        with pytest.raises(click.exceptions.Exit) as exc_info:
-            wait_for_package_sync(
-                ctx=ctx,
-                opts=opts,
-                owner="bart-demo-org-terraform",
-                repo="eng-13978-cli-repro",
-                slug="eng-13978-repro-100-alpha4tgz",
-                wait_interval=1.0,
-                skip_errors=False,
-                attempts=1,
-            )
+        wait_for_package_sync(
+            ctx=ctx,
+            opts=opts,
+            owner="bart-demo-org-terraform",
+            repo="eng-13978-cli-repro",
+            slug="eng-13978-repro-100-alpha4tgz",
+            wait_interval=1.0,
+            skip_errors=False,
+            attempts=1,
+        )
 
     assert exc_info.value.exit_code == 1
     captured = capsys.readouterr()

cloudsmith_cli/core/cache_utils.py

@@ -7,7 +7,9 @@
 import os
 import tempfile
 from collections.abc import Callable
-from typing import Any
+from typing import Any, Literal
+
+import toml
 
 
 def _atomic_write_text(dest: str, text: str, *, mode: int = 0o600) -> None:
@@ -44,13 +46,14 @@ def atomic_write_json(path: str | os.PathLike, data: Any, *, mode: int = 0o600)
     _atomic_write_text(dest, json.dumps(data), mode=mode)
 
 
-def merge_json_file(
+def merge_config_file(
     path: str | os.PathLike,
     mutate: Callable[[dict], None],
     *,
     backup: bool = True,
     dry_run: bool = False,
     mode: int = 0o600,
+    format: Literal["json", "toml"] = "json",
 ) -> bool:
     """Read a JSON object file, apply *mutate* in place, and atomically write it back.
 
@@ -69,6 +72,8 @@ def merge_json_file(
         **no** writes (no temp file, no ``.bak``, no replace).
     mode:
         File-permission bits for the written file (default ``0o600``).
+    format:
+        Format to read/write from
 
     Returns
     -------
@@ -110,13 +115,21 @@ def merge_json_file(
     # ------------------------------------------------------------------
     data: dict = {}
     if existing_text:
-        try:
-            parsed = json.loads(existing_text)
-            if isinstance(parsed, dict):
-                data = parsed
-        except (json.JSONDecodeError, ValueError):
-            pass
-
+        match format:
+            case "json":
+                try:
+                    parsed = json.loads(existing_text)
+                    if isinstance(parsed, dict):
+                        data = parsed
+                except (json.JSONDecodeError, ValueError):
+                    pass
+            case "toml":
+                try:
+                    parsed = toml.loads(existing_text)
+                    if isinstance(parsed, dict):
+                        data = parsed
+                except (toml.TomlDecodeError, ValueError):
+                    pass
     # ------------------------------------------------------------------
     # 3. Mutate in place
     # ------------------------------------------------------------------
@@ -125,7 +138,11 @@ def merge_json_file(
     # ------------------------------------------------------------------
     # 4. Stable serialisation + change detection
     # ------------------------------------------------------------------
-    new_text = json.dumps(data, indent=2, ensure_ascii=False) + "\n"
+    new_text = (
+        json.dumps(data, indent=2, ensure_ascii=False) + "\n"
+        if format == "json"
+        else toml.dumps(data)
+    )
 
     if existing_text is not None:
         # Normalise existing content for comparison: if the file already has

cloudsmith_cli/core/tests/test_aws_detector.py

@@ -106,11 +106,11 @@ def _get_token_sts_region(env):
             session = boto3.Session(
                 aws_access_key_id="test", aws_secret_access_key="test"
             )
-            with mock.patch.object(detector, "_session", session):
-                with mock.patch.object(
-                    session, "client", return_value=fake_sts
-                ) as client:
-                    assert detector.get_token() == "jwt"
+            with (
+                mock.patch.object(detector, "_session", session),
+                mock.patch.object(session, "client", return_value=fake_sts) as client,
+            ):
+                assert detector.get_token() == "jwt"
         (service_name,), call_kwargs = client.call_args
         assert service_name == "sts"
         return call_kwargs["region_name"]