diff --git a/CHANGELOG.md b/CHANGELOG.md index 3b314736..c08c2ba0 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added +- Added a Cargo credential provider for Cloudsmith registries. `cloudsmith credential-helper install cargo` installs a `cargo-credential-cloudsmith` launcher binary and registers it in `$CARGO_HOME/config.toml`, so Cargo authenticates to Cloudsmith registries automatically using your existing CLI credentials — no `cargo login` and no token in `credentials.toml`. `cloudsmith credential-helper cargo` speaks Cargo's [credential provider protocol](https://doc.rust-lang.org/cargo/reference/credential-provider-protocol.html): a newline-delimited JSON exchange that answers `get` with the resolved token, and answers a registry that is not a Cloudsmith one with `url-not-supported` so Cargo falls through to the next configured provider — registering globally cannot break authentication to crates.io. The provider is appended to `registry.global-credential-providers` (keeping `cargo:token` as the fallback) and pinned on any `[registries.*]` entry whose index points at a known Cloudsmith Cargo host. Custom Cloudsmith registry domains are discovered via the API and cached locally; add extra hostnames with `--domain` (repeatable), disable discovery with `--no-discover`, or preview changes with `--dry-run`. Manage installed helpers with `cloudsmith credential-helper uninstall cargo` and `cloudsmith credential-helper list`. - Added Nix package and upstream support. Use `cloudsmith push nix` to upload Nix packages and `cloudsmith upstream nix` to manage Nix channel upstreams. - Added a pnpm credential helper. `cloudsmith credential-helper install pnpm` registers `pnpm-credential-cloudsmith` in the user-level `.npmrc`, using existing CLI credentials for Cloudsmith registries. It supports custom-domain discovery, additional `--domain` values, `--no-discover`, `--dry-run`, listing, and uninstalling. - Added `CLOUDSMITH_KEYRING_FILE_PATH` and `CLOUDSMITH_KEYRING_DIR` to relocate tokens stored by the bundled file-based keyring backends. An explicit file path takes precedence over the directory, and `KEYRING_PROPERTY_FILE_PATH` takes precedence over its Cloudsmith alias. diff --git a/cloudsmith_cli/cli/commands/credential_helper/__init__.py b/cloudsmith_cli/cli/commands/credential_helper/__init__.py index c1bf2db5..66c4121c 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/__init__.py +++ b/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 @@ -31,6 +32,9 @@ def credential_helper(): # Install pnpm credential helper $ cloudsmith credential-helper install pnpm + # Install cargo credential helper + $ cloudsmith credential-helper install cargo + # Test Docker credential helper directly $ echo "docker.cloudsmith.io" | cloudsmith credential-helper docker @@ -45,5 +49,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") diff --git a/cloudsmith_cli/cli/commands/credential_helper/cargo.py b/cloudsmith_cli/cli/commands/credential_helper/cargo.py new file mode 100644 index 00000000..f40dd2d4 --- /dev/null +++ b/cloudsmith_cli/cli/commands/credential_helper/cargo.py @@ -0,0 +1,93 @@ +# Copyright 2026 Cloudsmith Ltd +""" +Cargo credential provider command. + +Implements the Cargo credential provider protocol for Cloudsmith registries. + +See: https://doc.rust-lang.org/cargo/reference/credential-provider-protocol.html +""" + +import sys + +import click + +from ....credential_helpers.cargo import execute +from ...decorators import common_api_auth_options, resolve_credentials + + +@click.command(context_settings={"ignore_unknown_options": True}) +@click.option( + "--cargo-plugin", + is_flag=True, + default=False, + hidden=True, + help="Passed by Cargo when invoking this command as a credential provider.", +) +@click.argument("provider_args", nargs=-1, type=click.UNPROCESSED) +@common_api_auth_options +@resolve_credentials +def cargo(opts, cargo_plugin, provider_args): # pylint: disable=unused-argument + """ + Cargo credential provider for Cloudsmith registries. + + Speaks the Cargo credential provider protocol: a newline-delimited JSON + conversation on stdin/stdout, starting with a hello message that announces + the supported protocol versions, then one response per request. + + Provides credentials for all Cloudsmith Cargo 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). + + A registry that is not a Cloudsmith one is answered with + ``url-not-supported`` so Cargo falls through to the next configured + credential provider. ``cargo login``/``cargo logout`` are answered with + ``operation-not-supported``: credentials come from the Cloudsmith CLI's own + provider chain, so there is nothing to store or clear. + + \b + Input (stdin): + One JSON request per line, e.g. + {"v":1,"kind":"get","operation":"read", + "registry":{"index-url":"sparse+https://cargo.cloudsmith.io/org/repo/"}} + + \b + Output (stdout): + {"v":[1]} + {"Ok":{"kind":"get","token":"","cache":"session", + "operation_independent":true}} + + \b + Exit codes: + 0: Session completed + 1: No credentials available, or the session broke down + + \b + Examples: + # Manual testing + $ echo '{"v":1,"kind":"get","operation":"read","registry":{"index-url":"sparse+https://cargo.cloudsmith.io/org/repo/"}}' \\ + | cloudsmith credential-helper cargo + + # Called by Cargo via the launcher + $ cargo-credential-cloudsmith --cargo-plugin + + \b + Environment variables: + CLOUDSMITH_API_KEY: API key for authentication (optional) + CLOUDSMITH_ORG: Organisation slug (required for custom domain support) + """ + # `provider_args` collects the extra arguments Cargo appends from the + # credential-provider config entry. This provider takes no configuration + # of its own, so they are accepted and ignored rather than rejected — an + # unknown-option error would surface as an authentication failure. + exit_code, stderr = execute( + sys.stdin, + sys.stdout, + credential=opts.credential, + api_host=opts.api_host, + org=opts.org, + ) + + if stderr is not None: + click.echo(stderr, err=True) + sys.exit(exit_code) diff --git a/cloudsmith_cli/cli/commands/credential_helper/manage.py b/cloudsmith_cli/cli/commands/credential_helper/manage.py index 2eaa94ef..f2745e3a 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/manage.py +++ b/cloudsmith_cli/cli/commands/credential_helper/manage.py @@ -12,6 +12,7 @@ import click +from cloudsmith_cli.credential_helpers.cargo.installer import CargoInstaller from cloudsmith_cli.credential_helpers.generic import PartialInstallError from cloudsmith_cli.credential_helpers.pnpm.installer import PNPMInstaller @@ -31,6 +32,7 @@ _INSTALLERS: dict[str, type] = { "docker": DockerInstaller, "pnpm": PNPMInstaller, + "cargo": CargoInstaller, } diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py index fd1c0a0f..9723032d 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py @@ -8,11 +8,11 @@ import stat import sys from pathlib import Path +from typing import TYPE_CHECKING from unittest.mock import patch import click.testing import pytest -from _pytest.monkeypatch import MonkeyPatch from cloudsmith_cli.credential_helpers.generic import PartialInstallError from cloudsmith_cli.credential_helpers.pnpm.installer import PNPMInstaller @@ -30,6 +30,9 @@ write_launcher, ) +if TYPE_CHECKING: + from _pytest.monkeypatch import MonkeyPatch + # --------------------------------------------------------------------------- # Fixtures # --------------------------------------------------------------------------- diff --git a/cloudsmith_cli/cli/tests/commands/test_mcp.py b/cloudsmith_cli/cli/tests/commands/test_mcp.py index 53913469..72a96f5b 100644 --- a/cloudsmith_cli/cli/tests/commands/test_mcp.py +++ b/cloudsmith_cli/cli/tests/commands/test_mcp.py @@ -5,7 +5,6 @@ from pathlib import Path from unittest.mock import patch -import cloudsmith_api import pytest from ....cli.commands.mcp import ( @@ -291,6 +290,8 @@ def test_server_generates_tools_from_openapi_spec(self): } } + import cloudsmith_api + # Create API config api_config = cloudsmith_api.Configuration() api_config.host = "https://api.cloudsmith.io" @@ -349,6 +350,8 @@ def test_server_respects_tool_filtering(self): } } + import cloudsmith_api + api_config = cloudsmith_api.Configuration() api_config.host = "https://api.cloudsmith.io" api_config.api_key = {"X-Api-Key": "test-key"} diff --git a/cloudsmith_cli/cli/tests/test_credential_helper_cargo.py b/cloudsmith_cli/cli/tests/test_credential_helper_cargo.py new file mode 100644 index 00000000..5ea6a4ae --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_credential_helper_cargo.py @@ -0,0 +1,667 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the `cloudsmith credential-helper cargo` command and installer.""" + +from __future__ import annotations + +import io +import json +from unittest.mock import patch + +import click.testing +import pytest +import tomlkit + +from ...core.credentials.models import CredentialResult +from ...credential_helpers.backends import BackendKind +from ...credential_helpers.cargo.installer import CargoInstaller +from ...credential_helpers.cargo.runtime import ( + _REFUSAL_MESSAGE, + SUPPORTED_VERSIONS, + execute, + get_credentials, + handle_request, + hello, +) +from ..commands.credential_helper.cargo import cargo + +CLOUDSMITH_INDEX = "sparse+https://cargo.cloudsmith.io/acme/repo/" +CRATES_IO_INDEX = "sparse+https://index.crates.io/" +PROVIDER = CargoInstaller.PROVIDER_VALUE +TOKEN_PROVIDER = CargoInstaller.TOKEN_PROVIDER + + +@pytest.fixture() +def runner(): + """Return a CliRunner.""" + return click.testing.CliRunner() + + +@pytest.fixture() +def credential(): + """Return a resolved credential.""" + return CredentialResult(api_key="k_abc", source_name="test") + + +def _request(**overrides) -> dict: + """Build a protocol-valid `get`/`read` request, overridden as needed.""" + request = { + "v": 1, + "kind": "get", + "operation": "read", + "registry": {"index-url": CLOUDSMITH_INDEX, "name": "acme"}, + "args": [], + } + request.update(overrides) + return request + + +def _session(*requests, credential=None, org=None): + """Run execute() over *requests* and return (code, stderr, messages).""" + stdin = io.StringIO("".join(json.dumps(r) + "\n" for r in requests)) + stdout = io.StringIO() + code, stderr = execute(stdin, stdout, credential=credential, org=org) + messages = [json.loads(line) for line in stdout.getvalue().splitlines()] + return code, stderr, messages + + +# --------------------------------------------------------------------------- +# 1. Handshake +# --------------------------------------------------------------------------- + + +def test_hello_announces_protocol_version_1(): + """The hello message lists the supported protocol versions as a JSON array.""" + assert hello() == {"v": [1]} + assert SUPPORTED_VERSIONS == (1,) + + +def test_hello_is_written_before_any_request_is_read(credential): + """Cargo blocks on the hello line, so it must be flushed before reading stdin.""" + + class AssertHelloFirst(io.StringIO): + """A stdin that asserts the hello line is already on stdout.""" + + def __iter__(self): + assert stdout.getvalue() == json.dumps(hello()) + "\n" + return iter([]) + + stdout = io.StringIO() + code, stderr = execute(AssertHelloFirst(""), stdout, credential=credential) + + assert (code, stderr) == (0, None) + assert stdout.getvalue() == json.dumps(hello()) + "\n" + + +def test_empty_session_is_not_an_error(credential): + """Cargo closing stdin without sending a request exits cleanly.""" + code, stderr, messages = _session(credential=credential) + + assert (code, stderr) == (0, None) + assert messages == [hello()] + + +# --------------------------------------------------------------------------- +# 2. get — the credential path +# --------------------------------------------------------------------------- + + +def test_get_returns_token_for_cloudsmith_registry(credential): + """A Cloudsmith index URL yields an Ok/get response carrying the token.""" + code, stderr, messages = _session(_request(), credential=credential) + + assert (code, stderr) == (0, None) + assert messages[0] == hello() + assert messages[1] == { + "Ok": { + "kind": "get", + "token": "k_abc", + "cache": "session", + "operation_independent": True, + } + } + + +@pytest.mark.parametrize("operation", ["read", "publish", "yank", "owners"]) +def test_get_serves_every_operation(operation, credential): + """The token is operation-independent, so every `get` operation is served.""" + request = _request(operation=operation, name="sample", vers="0.1.0") + _, _, messages = _session(request, credential=credential) + + assert messages[1]["Ok"]["token"] == "k_abc" + + +def test_get_uses_the_cargo_backend_kind_for_custom_domains(credential): + """Custom-domain matching is scoped to Cargo-backed domains.""" + with patch( + "cloudsmith_cli.credential_helpers.cargo.runtime.is_cloudsmith_domain", + return_value=True, + ) as mock_check: + _session( + _request(registry={"index-url": "sparse+https://crates.acme.com/"}), + credential=credential, + org="acme", + ) + + assert mock_check.call_args.kwargs["backend_kind"] is BackendKind.CARGO + assert mock_check.call_args.kwargs["org"] == "acme" + + +def test_index_url_keeps_its_sparse_prefix_out_of_the_host_match(credential): + """Cargo's `sparse+` prefix and the repo path don't defeat the host check.""" + assert ( + get_credentials(CLOUDSMITH_INDEX, credential=credential, org="acme") == "k_abc" + ) + + +# --------------------------------------------------------------------------- +# 3. get — the refusal paths, which Cargo distinguishes +# --------------------------------------------------------------------------- + + +def test_get_defers_to_the_next_provider_for_a_foreign_registry(credential): + """crates.io is answered url-not-supported so Cargo falls through, exit 0.""" + code, stderr, messages = _session( + _request(registry={"index-url": CRATES_IO_INDEX}), credential=credential + ) + + assert messages[1] == {"Err": {"kind": "url-not-supported"}} + # Not our registry is not an error: installing globally must not break + # authentication to crates.io. + assert (code, stderr) == (0, None) + + +def test_get_reports_not_found_when_no_credential_resolves(): + """A Cloudsmith registry with no token is not-found, exit 1 with a hint.""" + code, stderr, messages = _session(_request(), credential=None) + + assert messages[1] == {"Err": {"kind": "not-found"}} + assert code == 1 + assert stderr == _REFUSAL_MESSAGE + + +def test_get_reports_not_found_for_a_credential_without_an_api_key(): + """An empty api_key is treated as no credential at all.""" + empty = CredentialResult(api_key="", source_name="test") + code, _, messages = _session(_request(), credential=empty) + + assert messages[1] == {"Err": {"kind": "not-found"}} + assert code == 1 + + +@pytest.mark.parametrize( + "registry", + [ + None, + "cargo.cloudsmith.io", + {}, + {"name": "acme"}, + {"index-url": ""}, + ], +) +def test_get_without_a_usable_index_url_is_an_other_error(registry, credential): + """A request we cannot interpret is reported as `other`, with a message.""" + _, _, messages = _session(_request(registry=registry), credential=credential) + + assert messages[1]["Err"]["kind"] == "other" + assert messages[1]["Err"]["message"] + + +# --------------------------------------------------------------------------- +# 4. login / logout / unknown kinds +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("kind", ["login", "logout", "frobnicate", None]) +def test_unsupported_kinds_are_reported_as_operation_not_supported(kind, credential): + """Credentials come from the CLI's chain: there is nothing to store or clear.""" + request = _request(kind=kind, token="k_new", **{"login-url": "https://example.com"}) + code, stderr, messages = _session(request, credential=credential) + + assert messages[1] == {"Err": {"kind": "operation-not-supported"}} + assert (code, stderr) == (0, None) + + +# --------------------------------------------------------------------------- +# 5. Malformed input and version negotiation +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("version", [None, 0, 2, "1"]) +def test_unsupported_protocol_version_is_rejected(version, credential): + """Only the versions announced in the hello message are answered.""" + _, _, messages = _session(_request(v=version), credential=credential) + + assert messages[1]["Err"]["kind"] == "other" + assert "Unsupported protocol version" in messages[1]["Err"]["message"] + + +@pytest.mark.parametrize("line", ["{not json", "[]", '"a string"', "null", "3"]) +def test_malformed_request_lines_are_answered_not_crashed(line, credential): + """A line that isn't a JSON object gets an `other` error, and the loop lives on.""" + stdin = io.StringIO(line + "\n" + json.dumps(_request()) + "\n") + stdout = io.StringIO() + + code, stderr = execute(stdin, stdout, credential=credential) + messages = [json.loads(m) for m in stdout.getvalue().splitlines()] + + assert (code, stderr) == (0, None) + assert messages[1]["Err"]["kind"] == "other" + # The following well-formed request is still served. + assert messages[2]["Ok"]["token"] == "k_abc" + + +def test_blank_lines_are_skipped(credential): + """Blank lines produce no response message.""" + stdin = io.StringIO("\n\n" + json.dumps(_request()) + "\n\n") + stdout = io.StringIO() + + execute(stdin, stdout, credential=credential) + + assert len(stdout.getvalue().splitlines()) == 2 + + +def test_multiple_requests_are_answered_in_one_session(credential): + """One response per request, in order, on a single long-lived session.""" + code, _, messages = _session( + _request(), + _request(registry={"index-url": CRATES_IO_INDEX}), + _request(kind="login"), + credential=credential, + ) + + assert code == 0 + assert "Ok" in messages[1] + assert messages[2]["Err"]["kind"] == "url-not-supported" + assert messages[3]["Err"]["kind"] == "operation-not-supported" + + +def test_transport_failure_degrades_to_a_clean_exit(credential): + """A broken pipe on stdout must not surface as a traceback.""" + + class BrokenPipe(io.StringIO): + def write(self, _s): + raise OSError("broken pipe") + + code, stderr = execute(io.StringIO(""), BrokenPipe(), credential=credential) + + assert code == 1 + assert stderr == _REFUSAL_MESSAGE + + +def test_domain_lookup_failure_degrades_to_a_clean_exit(credential): + """A network error during custom-domain discovery must not crash Cargo.""" + with patch( + "cloudsmith_cli.credential_helpers.cargo.runtime.is_cloudsmith_domain", + side_effect=RuntimeError("boom"), + ): + code, stderr, messages = _session(_request(), credential=credential) + + assert (code, stderr) == (1, _REFUSAL_MESSAGE) + assert messages == [hello()] + + +def test_handle_request_never_raises_on_a_non_dict(): + """handle_request is total over decoded JSON values.""" + assert handle_request(["not", "a", "dict"])["Err"]["kind"] == "other" + + +# --------------------------------------------------------------------------- +# 6. CLI wiring +# --------------------------------------------------------------------------- + + +def test_cli_speaks_the_protocol_on_stdin_and_stdout(runner): + """The click shim wires stdin/stdout through to a full protocol exchange.""" + result = runner.invoke( + cargo, + args=["-k", "k_abc"], + input=json.dumps(_request()) + "\n", + catch_exceptions=False, + ) + + messages = [json.loads(line) for line in result.stdout.splitlines()] + assert result.exit_code == 0 + assert messages[0] == hello() + assert messages[1]["Ok"]["token"] == "k_abc" + + +def test_cli_accepts_the_cargo_plugin_flag_and_extra_provider_args(runner): + """Cargo passes --cargo-plugin plus any config args; neither may be an error.""" + result = runner.invoke( + cargo, + args=["-k", "k_abc", "--cargo-plugin", "--some-config-arg", "value"], + input=json.dumps(_request()) + "\n", + catch_exceptions=False, + ) + + messages = [json.loads(line) for line in result.stdout.splitlines()] + assert result.exit_code == 0 + assert messages[1]["Ok"]["token"] == "k_abc" + + +def test_cli_exits_non_zero_with_a_hint_when_no_credential_resolves(runner): + """A refused `get` still answers Cargo in-band, then exits 1 with a hint.""" + with patch( + "cloudsmith_cli.credential_helpers.cargo.runtime.is_cloudsmith_domain", + return_value=True, + ): + result = runner.invoke( + cargo, + args=[], + input=json.dumps(_request()) + "\n", + env={"CLOUDSMITH_API_KEY": ""}, + catch_exceptions=False, + ) + + assert result.exit_code == 1 + assert '{"Err": {"kind": "not-found"}}' in result.stdout + + +# --------------------------------------------------------------------------- +# 7. Installer +# --------------------------------------------------------------------------- + + +@pytest.fixture() +def cargo_home(tmp_path, monkeypatch): + """Point CARGO_HOME at a temporary directory and return its config.toml.""" + home = tmp_path / ".cargo" + monkeypatch.setenv("CARGO_HOME", str(home)) + return home / "config.toml" + + +@pytest.fixture() +def bin_dir(tmp_path, monkeypatch): + """Return a launcher directory that is on PATH.""" + path = tmp_path / "bin" + monkeypatch.setenv("PATH", str(path)) + return path + + +def test_installer_registers_the_provider_globally(cargo_home, bin_dir): + """install appends the provider to global-credential-providers, cargo:token kept.""" + installer = CargoInstaller() + actions = installer.install(bin_dir=str(bin_dir), discover=False) + + config = tomlkit.loads(cargo_home.read_text()) + # Cargo tries providers last-to-first, so ours must come last to be tried first. + assert config["registry"]["global-credential-providers"] == [ + TOKEN_PROVIDER, + PROVIDER, + ] + assert (bin_dir / "cargo-credential-cloudsmith").exists() + assert not any(a.startswith("WARNING") for a in actions) + + +def test_installer_writes_a_launcher_cargo_can_discover(cargo_home, bin_dir): + """The launcher name carries Cargo's required prefix and execs the CLI.""" + CargoInstaller().install(bin_dir=str(bin_dir), discover=False) + + launcher = bin_dir / "cargo-credential-cloudsmith" + assert launcher.name.startswith("cargo-credential-") + assert launcher.read_text() == ( + '#!/bin/sh\nexec cloudsmith credential-helper cargo "$@"\n' + ) + + +def test_installer_preserves_foreign_config_and_provider_order(cargo_home, bin_dir): + """Existing config, including other providers, survives the merge.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text( + tomlkit.dumps( + { + "build": {"jobs": 4}, + "registry": { + "global-credential-providers": [ + TOKEN_PROVIDER, + "cargo:macos-keychain", + ] + }, + } + ) + ) + + CargoInstaller().install(bin_dir=str(bin_dir), discover=False) + + config = tomlkit.loads(cargo_home.read_text()) + assert config["build"] == {"jobs": 4} + assert config["registry"]["global-credential-providers"] == [ + TOKEN_PROVIDER, + "cargo:macos-keychain", + PROVIDER, + ] + + +def test_installer_adds_the_token_provider_when_the_key_is_absent(cargo_home, bin_dir): + """Setting the key overrides Cargo's default, so cargo:token is carried over.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text(tomlkit.dumps({"registry": {"default": "acme"}})) + + CargoInstaller().install(bin_dir=str(bin_dir), discover=False) + + config = tomlkit.loads(cargo_home.read_text()) + assert config["registry"]["default"] == "acme" + assert TOKEN_PROVIDER in config["registry"]["global-credential-providers"] + + +def test_installer_is_idempotent(cargo_home, bin_dir): + """A second install neither duplicates the provider nor rewrites the file.""" + installer = CargoInstaller() + installer.install(bin_dir=str(bin_dir), discover=False) + mtime_before = cargo_home.stat().st_mtime + + actions = installer.install(bin_dir=str(bin_dir), discover=False) + + assert cargo_home.stat().st_mtime == mtime_before + assert tomlkit.loads(cargo_home.read_text())["registry"][ + "global-credential-providers" + ] == [TOKEN_PROVIDER, PROVIDER] + assert any("already up to date" in a for a in actions) + + +def test_installer_pins_the_provider_on_matching_named_registries(cargo_home, bin_dir): + """A per-registry credential-provider shadows the global list, so pin it too.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text( + tomlkit.dumps( + { + "registries": { + "acme": {"index": CLOUDSMITH_INDEX}, + "custom": {"index": "sparse+https://crates.acme.com/index/"}, + "crates-io-mirror": {"index": CRATES_IO_INDEX}, + } + } + ) + ) + + actions = CargoInstaller().install( + bin_dir=str(bin_dir), + domains=("crates.acme.com",), + discover=False, + ) + + registries = tomlkit.loads(cargo_home.read_text())["registries"] + assert registries["acme"]["credential-provider"] == PROVIDER + # Reached via --domain + assert registries["custom"]["credential-provider"] == PROVIDER + # A foreign registry is left for its own provider + assert "credential-provider" not in registries["crates-io-mirror"] + assert any("registries.acme.credential-provider" in a for a in actions) + + +def test_installer_dry_run_writes_nothing(cargo_home, bin_dir): + """--dry-run reports the plan and touches neither the config nor PATH.""" + actions = CargoInstaller().install(bin_dir=str(bin_dir), dry_run=True) + + assert not cargo_home.exists() + assert not (bin_dir / "cargo-credential-cloudsmith").exists() + assert any("would write launcher" in a for a in actions) + assert any("global-credential-providers" in a for a in actions) + assert any("skipped custom-domain auto-discovery" in a for a in actions) + + +def test_installer_discovers_custom_domains(cargo_home, bin_dir, credential): + """Discovered Cargo domains widen which named registries get pinned.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text( + tomlkit.dumps( + {"registries": {"acme": {"index": "sparse+https://crates.acme.com/"}}} + ) + ) + + with patch( + "cloudsmith_cli.credential_helpers.cargo.installer.get_format_domains", + return_value=["crates.acme.com"], + ) as mock_discover: + CargoInstaller().install( + bin_dir=str(bin_dir), org="acme", credential=credential + ) + + assert mock_discover.call_args.args[1] is BackendKind.CARGO + registries = tomlkit.loads(cargo_home.read_text())["registries"] + assert registries["acme"]["credential-provider"] == PROVIDER + + +def test_installer_survives_discovery_failure(cargo_home, bin_dir, credential): + """Discovery is best-effort: the global registration still happens.""" + with patch( + "cloudsmith_cli.credential_helpers.cargo.installer.get_format_domains", + side_effect=RuntimeError("boom"), + ): + actions = CargoInstaller().install( + bin_dir=str(bin_dir), org="acme", credential=credential + ) + + config = tomlkit.loads(cargo_home.read_text()) + assert PROVIDER in config["registry"]["global-credential-providers"] + assert any("auto-discovery failed" in a for a in actions) + + +def test_installer_warns_when_the_launcher_is_not_on_path( + cargo_home, tmp_path, monkeypatch +): + """Cargo resolves a bare provider name through PATH, so warn when it can't.""" + monkeypatch.setenv("PATH", "") + + actions = CargoInstaller().install(bin_dir=str(tmp_path / "bin"), discover=False) + + assert any("is not on PATH" in a for a in actions) + + +def test_installer_uninstall_restores_the_previous_config(cargo_home, bin_dir): + """uninstall removes the launcher, the global entry and the pinned entries.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text( + tomlkit.dumps( + { + "build": {"jobs": 4}, + "registries": {"acme": {"index": CLOUDSMITH_INDEX}}, + } + ) + ) + installer = CargoInstaller() + installer.install(bin_dir=str(bin_dir), discover=False) + + actions = installer.uninstall(bin_dir=str(bin_dir)) + + config = tomlkit.loads(cargo_home.read_text()) + assert not (bin_dir / "cargo-credential-cloudsmith").exists() + # Only cargo:token would be left, which is Cargo's default — so the key goes. + assert "registry" not in config + assert "credential-provider" not in config["registries"]["acme"] + assert config["registries"]["acme"]["index"] == CLOUDSMITH_INDEX + assert config["build"] == {"jobs": 4} + assert any("removed launcher" in a for a in actions) + + +def test_installer_uninstall_keeps_other_providers(cargo_home, bin_dir): + """Only our entry is removed from a list the user has customised.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text( + tomlkit.dumps( + { + "registry": { + "global-credential-providers": [ + TOKEN_PROVIDER, + "cargo:libsecret", + PROVIDER, + ] + } + } + ) + ) + + CargoInstaller().uninstall(bin_dir=str(bin_dir)) + + config = tomlkit.loads(cargo_home.read_text()) + assert config["registry"]["global-credential-providers"] == [ + TOKEN_PROVIDER, + "cargo:libsecret", + ] + + +def test_installer_uninstall_is_a_no_op_when_not_installed(cargo_home, bin_dir): + """Uninstalling a helper that was never installed reports nothing to do.""" + actions = CargoInstaller().uninstall(bin_dir=str(bin_dir)) + + # Merging into an absent config would leave an empty config.toml behind. + assert not cargo_home.exists() + assert any("nothing to remove" in a for a in actions) + assert any("entries to remove" in a for a in actions) + + +def test_installer_uninstall_dry_run_writes_nothing(cargo_home, bin_dir): + """--dry-run on uninstall leaves the install in place.""" + installer = CargoInstaller() + installer.install(bin_dir=str(bin_dir), discover=False) + + actions = installer.uninstall(bin_dir=str(bin_dir), dry_run=True) + + config = tomlkit.loads(cargo_home.read_text()) + assert PROVIDER in config["registry"]["global-credential-providers"] + assert (bin_dir / "cargo-credential-cloudsmith").exists() + assert any("would remove launcher" in a for a in actions) + + +def test_installer_status_type_contract(cargo_home, bin_dir): + """status reports the launcher path and where the provider is registered.""" + installer = CargoInstaller() + + with patch( + "cloudsmith_cli.credential_helpers.cargo.installer.resolve_bin_dir", + return_value=bin_dir, + ): + assert installer.status() == {"launcher": None, "hosts": []} + + installer.install(bin_dir=str(bin_dir), discover=False) + status = installer.status() + + assert status["launcher"].endswith("cargo-credential-cloudsmith") + assert status["hosts"] == ["all registries (global-credential-providers)"] + + +def test_installer_status_lists_pinned_registry_hosts(cargo_home, bin_dir): + """A registry pinned to the provider is reported by its index hostname.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text( + tomlkit.dumps( + { + "registries": { + "acme": { + "index": CLOUDSMITH_INDEX, + "credential-provider": PROVIDER, + } + } + } + ) + ) + + assert CargoInstaller().status()["hosts"] == ["cargo.cloudsmith.io"] + + +def test_installer_status_survives_a_malformed_config(cargo_home): + """A config.toml Cargo itself would reject must not crash `list`.""" + cargo_home.parent.mkdir(parents=True) + cargo_home.write_text("this is [not valid toml") + + assert CargoInstaller().status()["hosts"] == [] diff --git a/cloudsmith_cli/cli/tests/test_exceptions.py b/cloudsmith_cli/cli/tests/test_exceptions.py index 588e479a..56455ffc 100644 --- a/cloudsmith_cli/cli/tests/test_exceptions.py +++ b/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." diff --git a/cloudsmith_cli/cli/tests/test_push.py b/cloudsmith_cli/cli/tests/test_push.py index 1b6c3fe6..f5029705 100644 --- a/cloudsmith_cli/cli/tests/test_push.py +++ b/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() diff --git a/cloudsmith_cli/cli/tests/test_saml.py b/cloudsmith_cli/cli/tests/test_saml.py index 0f219414..5d7e7b6a 100644 --- a/cloudsmith_cli/cli/tests/test_saml.py +++ b/cloudsmith_cli/cli/tests/test_saml.py @@ -1,7 +1,6 @@ from unittest.mock import MagicMock, patch import pytest -import requests from ...core.api.exceptions import ApiException from ..saml import exchange_2fa_token, get_idp_url, refresh_access_token @@ -15,6 +14,8 @@ def mock_response(): @pytest.fixture def mock_session(): + import requests + session = MagicMock(spec=requests.sessions.Session) return session @@ -38,6 +39,8 @@ def test_get_idp_url(self, mock_response, mock_session): ) def test_get_idp_url_with_request_error(self, mock_response, mock_session): + import requests + mock_session.get.return_value = mock_response mock_response.status_code = 500 mock_response.headers = {"foo": "bar"} @@ -57,6 +60,8 @@ def test_get_idp_url_with_request_error(self, mock_response, mock_session): ) def test_error_carries_the_api_detail(self, mock_response, mock_session): + import requests + """Verify the API's own explanation reaches the message the user reads.""" mock_session.post.return_value = mock_response mock_response.status_code = 401 @@ -76,6 +81,8 @@ def test_error_carries_the_api_detail(self, mock_response, mock_session): assert str(exc_info.value) == "401 - Your session has expired." def test_error_without_json_body_still_raises(self, mock_response, mock_session): + import requests + """Verify a non-JSON error body falls back to the status description.""" mock_session.post.return_value = mock_response mock_response.status_code = 502 @@ -118,6 +125,8 @@ def test_exchange_2fa_token(self, mock_response, mock_session): ) def test_exchange_2fa_token_with_request_error(self, mock_response, mock_session): + import requests + mock_session.post.return_value = mock_response mock_response.status_code = 500 mock_response.headers = {"foo": "bar"} @@ -171,6 +180,8 @@ def test_refresh_access_token(self, mock_response, mock_session): ) def test_refresh_access_token_with_request_error(self, mock_response, mock_session): + import requests + mock_session.post.return_value = mock_response mock_response.status_code = 500 mock_response.headers = {"foo": "bar"} diff --git a/cloudsmith_cli/cli/tests/test_startup_imports.py b/cloudsmith_cli/cli/tests/test_startup_imports.py index 5068d76f..122b688f 100644 --- a/cloudsmith_cli/cli/tests/test_startup_imports.py +++ b/cloudsmith_cli/cli/tests/test_startup_imports.py @@ -30,6 +30,7 @@ def modules_loaded_by_import(module_name="cloudsmith_cli.cli.commands.main"): [sys.executable, "-c", code], capture_output=True, text=True, + check=False, ) assert result.returncode == 0, result.stderr return json.loads(result.stdout) diff --git a/cloudsmith_cli/core/cache_utils.py b/cloudsmith_cli/core/cache_utils.py index b2e31300..1ee016e1 100644 --- a/cloudsmith_cli/core/cache_utils.py +++ b/cloudsmith_cli/core/cache_utils.py @@ -6,7 +6,10 @@ import json import os import tempfile -from typing import TYPE_CHECKING, Any +from collections.abc import Callable +from typing import TYPE_CHECKING, Any, Literal + +import tomlkit if TYPE_CHECKING: from collections.abc import Callable @@ -46,13 +49,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. @@ -71,6 +75,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. (JSON or TOML) Returns ------- @@ -111,14 +117,21 @@ def merge_json_file( # 2. Parse → dict (treat missing/empty/non-dict/malformed as {}) # ------------------------------------------------------------------ data: dict = {} + parsed: 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) + except (json.JSONDecodeError, ValueError): + pass + case "toml": + try: + parsed = tomlkit.loads(existing_text) + except (tomlkit.exceptions.ParseError, ValueError): + pass + if isinstance(parsed, dict) and parsed != {}: + data = parsed # ------------------------------------------------------------------ # 3. Mutate in place # ------------------------------------------------------------------ @@ -127,7 +140,14 @@ def merge_json_file( # ------------------------------------------------------------------ # 4. Stable serialisation + change detection # ------------------------------------------------------------------ - new_text = json.dumps(data, indent=2, ensure_ascii=False) + "\n" + new_text = "" + + if format == "json": + new_text = json.dumps(data, indent=2, ensure_ascii=False) + "\n" + elif format == "toml": + new_text = tomlkit.dumps(data) + else: + raise ValueError(f"Unsupported format {format!r}") if existing_text is not None: # Normalise existing content for comparison: if the file already has diff --git a/cloudsmith_cli/core/tests/test_aws_detector.py b/cloudsmith_cli/core/tests/test_aws_detector.py index 01dc2e9d..9cabd800 100644 --- a/cloudsmith_cli/core/tests/test_aws_detector.py +++ b/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"] diff --git a/cloudsmith_cli/core/tests/test_cache_utils.py b/cloudsmith_cli/core/tests/test_cache_utils.py index fcc7eed4..d660d025 100644 --- a/cloudsmith_cli/core/tests/test_cache_utils.py +++ b/cloudsmith_cli/core/tests/test_cache_utils.py @@ -7,7 +7,7 @@ import os import stat -from cloudsmith_cli.core.cache_utils import atomic_write_json, merge_json_file +from cloudsmith_cli.core.cache_utils import atomic_write_json, merge_config_file # --------------------------------------------------------------------------- # Helpers @@ -58,7 +58,7 @@ def test_overwrites_existing(self, tmp_path): # --------------------------------------------------------------------------- -# merge_json_file +# merge_config_file # --------------------------------------------------------------------------- @@ -83,7 +83,7 @@ def test_existing_keys_preserved(self, tmp_path): with open(path, "w", encoding="utf-8") as f: json.dump(initial, f) - changed = merge_json_file( + changed = merge_config_file( path, _add_cred_helper("docker.cloudsmith.io"), ) @@ -104,7 +104,7 @@ def test_key_order_not_sorted(self, tmp_path): def noop(data: dict) -> None: data["new_key"] = 3 - merge_json_file(path, noop) + merge_config_file(path, noop) text = _read_text(path) assert text.index('"zzz"') < text.index('"aaa"'), "Key order must be preserved" @@ -114,7 +114,7 @@ class TestMergeJsonFileCreatesMissingFile: def test_creates_file_when_missing(self, tmp_path): path = str(tmp_path / "subdir" / "config.json") - changed = merge_json_file(path, _add_cred_helper("docker.cloudsmith.io")) + changed = merge_config_file(path, _add_cred_helper("docker.cloudsmith.io")) assert changed is True assert os.path.exists(path) result = _read_json(path) @@ -123,18 +123,18 @@ def test_creates_file_when_missing(self, tmp_path): def test_creates_parent_directory(self, tmp_path): path = str(tmp_path / "missing_dir" / "config.json") assert not os.path.exists(os.path.dirname(path)) - merge_json_file(path, _add_cred_helper("x")) + merge_config_file(path, _add_cred_helper("x")) assert os.path.isdir(os.path.dirname(path)) def test_parent_dir_permissions(self, tmp_path): path = str(tmp_path / "newdir" / "config.json") - merge_json_file(path, _add_cred_helper("x")) + merge_config_file(path, _add_cred_helper("x")) parent_perms = _perms(os.path.dirname(path)) assert parent_perms == 0o700 def test_file_permissions_after_create(self, tmp_path): path = str(tmp_path / "newdir" / "config.json") - merge_json_file(path, _add_cred_helper("x")) + merge_config_file(path, _add_cred_helper("x")) assert _perms(path) == 0o600 @@ -147,7 +147,7 @@ def test_backup_created_on_change(self, tmp_path): with open(path, "w", encoding="utf-8") as f: json.dump(initial, f) - merge_json_file(path, _add_cred_helper("docker.cloudsmith.io")) + merge_config_file(path, _add_cred_helper("docker.cloudsmith.io")) bak_path = path + ".bak" assert os.path.exists(bak_path), ".bak file should exist after a change" @@ -155,7 +155,7 @@ def test_backup_created_on_change(self, tmp_path): def test_no_backup_when_file_missing(self, tmp_path): path = str(tmp_path / "config.json") - merge_json_file(path, _add_cred_helper("x")) + merge_config_file(path, _add_cred_helper("x")) assert not os.path.exists(path + ".bak") def test_no_backup_when_no_change(self, tmp_path): @@ -166,7 +166,7 @@ def test_no_backup_when_no_change(self, tmp_path): def noop_already_set(data: dict) -> None: data.setdefault("credHelpers", {})["x"] = "cloudsmith" - changed = merge_json_file(path, noop_already_set) + changed = merge_config_file(path, noop_already_set) assert changed is False assert not os.path.exists(path + ".bak") @@ -177,7 +177,7 @@ def test_backup_is_mode_0o600_regardless_of_source_perms(self, tmp_path): json.dump({"auths": {}}, f) os.chmod(path, 0o644) - merge_json_file(path, _add_cred_helper("docker.cloudsmith.io")) + merge_config_file(path, _add_cred_helper("docker.cloudsmith.io")) bak_path = path + ".bak" assert os.path.exists(bak_path), ".bak must be created" @@ -193,10 +193,10 @@ def test_idempotent_returns_false_second_call(self, tmp_path): path = str(tmp_path / "config.json") mutate = _add_cred_helper("docker.cloudsmith.io") - first = merge_json_file(path, mutate) + first = merge_config_file(path, mutate) assert first is True - second = merge_json_file(path, mutate) + second = merge_config_file(path, mutate) assert second is False def test_idempotent_no_overwrite_bak(self, tmp_path): @@ -206,12 +206,12 @@ def test_idempotent_no_overwrite_bak(self, tmp_path): json.dump(initial, f) mutate = _add_cred_helper("docker.cloudsmith.io") - merge_json_file(path, mutate) # first: changes file, writes .bak + merge_config_file(path, mutate) # first: changes file, writes .bak bak_path = path + ".bak" bak_mtime_after_first = os.path.getmtime(bak_path) - merge_json_file(path, mutate) # second: no change + merge_config_file(path, mutate) # second: no change bak_mtime_after_second = os.path.getmtime(bak_path) assert bak_mtime_after_first == bak_mtime_after_second, ( @@ -227,7 +227,7 @@ def test_dry_run_returns_true_when_would_change(self, tmp_path): with open(path, "w", encoding="utf-8") as f: json.dump({}, f) - result = merge_json_file(path, _add_cred_helper("x"), dry_run=True) + result = merge_config_file(path, _add_cred_helper("x"), dry_run=True) assert result is True def test_dry_run_file_unchanged(self, tmp_path): @@ -238,7 +238,7 @@ def test_dry_run_file_unchanged(self, tmp_path): f.write(json.dumps({"existing": True}, indent=2) + "\n") original_text = _read_text(path) - merge_json_file(path, _add_cred_helper("x"), dry_run=True) + merge_config_file(path, _add_cred_helper("x"), dry_run=True) assert _read_text(path) == original_text, "dry_run must not modify the file" @@ -247,7 +247,7 @@ def test_dry_run_no_bak_created(self, tmp_path): with open(path, "w", encoding="utf-8") as f: json.dump({"existing": True}, f) - merge_json_file(path, _add_cred_helper("x"), dry_run=True) + merge_config_file(path, _add_cred_helper("x"), dry_run=True) assert not os.path.exists(path + ".bak") def test_dry_run_returns_false_when_no_change(self, tmp_path): @@ -259,12 +259,12 @@ def test_dry_run_returns_false_when_no_change(self, tmp_path): def already_set(data: dict) -> None: data.setdefault("credHelpers", {})["x"] = "cloudsmith" - result = merge_json_file(path, already_set, dry_run=True) + result = merge_config_file(path, already_set, dry_run=True) assert result is False def test_dry_run_missing_file_no_creation(self, tmp_path): path = str(tmp_path / "ghost" / "config.json") - result = merge_json_file(path, _add_cred_helper("x"), dry_run=True) + result = merge_config_file(path, _add_cred_helper("x"), dry_run=True) assert result is True assert not os.path.exists(path) assert not os.path.exists(os.path.dirname(path)) @@ -278,7 +278,7 @@ def test_malformed_json_treated_as_empty(self, tmp_path): with open(path, "w", encoding="utf-8") as f: f.write("not json") - changed = merge_json_file(path, _add_cred_helper("docker.cloudsmith.io")) + changed = merge_config_file(path, _add_cred_helper("docker.cloudsmith.io")) assert changed is True result = _read_json(path) assert result == {"credHelpers": {"docker.cloudsmith.io": "cloudsmith"}} @@ -288,7 +288,7 @@ def test_empty_file_treated_as_empty_dict(self, tmp_path): with open(path, "w", encoding="utf-8"): pass # touch / create empty file - merge_json_file(path, _add_cred_helper("x")) + merge_config_file(path, _add_cred_helper("x")) result = _read_json(path) assert "credHelpers" in result @@ -297,7 +297,7 @@ def test_json_array_treated_as_empty_dict(self, tmp_path): with open(path, "w", encoding="utf-8") as f: json.dump([1, 2, 3], f) - merge_json_file(path, _add_cred_helper("x")) + merge_config_file(path, _add_cred_helper("x")) result = _read_json(path) assert isinstance(result, dict) assert "credHelpers" in result @@ -308,7 +308,7 @@ class TestMergeJsonFileStableSerialization: def test_output_format(self, tmp_path): path = str(tmp_path / "config.json") - merge_json_file(path, _add_cred_helper("docker.cloudsmith.io")) + merge_config_file(path, _add_cred_helper("docker.cloudsmith.io")) text = _read_text(path) expected = json.dumps( {"credHelpers": {"docker.cloudsmith.io": "cloudsmith"}}, @@ -319,7 +319,7 @@ def test_output_format(self, tmp_path): def test_trailing_newline(self, tmp_path): path = str(tmp_path / "config.json") - merge_json_file(path, _add_cred_helper("x")) + merge_config_file(path, _add_cred_helper("x")) text = _read_text(path) assert text.endswith("\n") @@ -330,7 +330,7 @@ def test_non_ascii_host_raw_utf8_not_escaped(self, tmp_path): mutate = _add_cred_helper(unicode_host) # First call: file is created (content changes → True) - first = merge_json_file(path, mutate) + first = merge_config_file(path, mutate) assert first is True # The written file must contain the raw Unicode character @@ -347,7 +347,7 @@ def test_non_ascii_host_raw_utf8_not_escaped(self, tmp_path): os.path.getmtime(bak_path) if os.path.exists(bak_path) else None ) - second = merge_json_file(path, mutate) + second = merge_config_file(path, mutate) assert second is False # .bak must not have been touched on the no-op call @@ -364,12 +364,12 @@ class TestMergeJsonFileReturnValue: def test_returns_true_on_actual_write(self, tmp_path): path = str(tmp_path / "config.json") - result = merge_json_file(path, _add_cred_helper("x")) + result = merge_config_file(path, _add_cred_helper("x")) assert result is True def test_returns_false_on_no_change(self, tmp_path): path = str(tmp_path / "config.json") mutate = _add_cred_helper("x") - merge_json_file(path, mutate) - result = merge_json_file(path, mutate) + merge_config_file(path, mutate) + result = merge_config_file(path, mutate) assert result is False diff --git a/cloudsmith_cli/core/tests/test_init.py b/cloudsmith_cli/core/tests/test_init.py index f0b691e6..81ea2311 100644 --- a/cloudsmith_cli/core/tests/test_init.py +++ b/cloudsmith_cli/core/tests/test_init.py @@ -1,10 +1,10 @@ -from cloudsmith_api import Configuration - from ..api.init import initialise_api class TestInitialiseApi: def setup_class(cls): # pylint: disable=no-self-argument + from cloudsmith_api import Configuration + # For the purposes of these tests, we need to explicitly call set_default(None) at the # outset because other tests in the suite may have called initialise_api() already. # Resetting Configuration._default to None here effectively reverts the @@ -36,6 +36,8 @@ def test_initialise_api_sets_cloudsmith_api_config_default(self): # Because Configuration._default is None, a newly-created instance of # cloudsmith_api.Configuration() should not have any other attributes than those # in the auto-generated swagger-codegen class declaration. + from cloudsmith_api import Configuration + new_config_before_initialise = Configuration() assert all( not hasattr(new_config_before_initialise, attr) @@ -92,6 +94,8 @@ def test_initialise_api_sets_api_key(self): def test_initialise_api_bearer_credential(self): """Verify bearer credential sets Authorization header, not X-Api-Key.""" + from cloudsmith_api import Configuration + from cloudsmith_cli.core.credentials.models import CredentialResult Configuration.set_default(None) @@ -107,6 +111,8 @@ def test_initialise_api_bearer_credential(self): def test_initialise_api_with_basic_auth_header(self): """Verify basic auth header is parsed into username and password.""" + from cloudsmith_api import Configuration + temp_config = Configuration() temp_config.username = "username" temp_config.password = "password" diff --git a/cloudsmith_cli/core/tests/test_keyring.py b/cloudsmith_cli/core/tests/test_keyring.py index 4fd96e04..485fe977 100644 --- a/cloudsmith_cli/core/tests/test_keyring.py +++ b/cloudsmith_cli/core/tests/test_keyring.py @@ -3,10 +3,8 @@ from datetime import datetime, timedelta, timezone from unittest.mock import ANY, patch -import keyring import pytest from freezegun import freeze_time -from keyring.errors import KeyringError from keyrings.cryptfile.cryptfile import CryptFileKeyring from ..keyring import ( @@ -32,24 +30,32 @@ def mock_get_user(): @pytest.fixture def mock_get_password(): + import keyring + with patch.object(keyring, "get_password") as get_password_mock: yield get_password_mock @pytest.fixture def mock_set_password(): + import keyring + with patch.object(keyring, "set_password") as set_password_mock: yield set_password_mock @pytest.fixture def mock_delete_password(): + import keyring + with patch.object(keyring, "delete_password") as delete_password_mock: yield delete_password_mock @pytest.fixture(autouse=True) def mock_get_keyring(): + import keyring + with patch.object(keyring, "get_keyring") as get_keyring_mock: yield get_keyring_mock @@ -75,6 +81,8 @@ def test_get_access_token(self, mock_get_user, mock_get_password): ) def test_get_access_token_when_error_raised(self, mock_get_user, mock_get_password): + from keyring.errors import KeyringError + mock_get_password.side_effect = KeyringError("A keyring error occurred") assert get_access_token(self.api_host) is None @@ -110,6 +118,8 @@ def test_get_refresh_attempted_at(self, mock_get_user, mock_get_password): def test_get_refresh_attempted_at_when_keyring_error_raised( self, mock_get_user, mock_get_password ): + from keyring.errors import KeyringError + mock_get_password.side_effect = KeyringError("A keyring error occurred") assert get_refresh_attempted_at(self.api_host) is None @@ -189,6 +199,8 @@ def test_get_refresh_token(self, mock_get_user, mock_get_password): def test_get_refresh_token_when_error_raised( self, mock_get_user, mock_get_password ): + from keyring.errors import KeyringError + mock_get_password.side_effect = KeyringError("A keyring error occurred") assert get_refresh_token(self.api_host) is None @@ -536,6 +548,8 @@ def _relocation_env(tmp_path): return env def _roundtrip_token(self, backend, env): + import keyring + with ( patch.dict(os.environ, env, clear=True), patch.object(keyring, "set_password", side_effect=backend.set_password), @@ -575,6 +589,8 @@ def test_file_path_alias_relocates_storage_before_unlock( class TestDeleteSsoTokens: """Tests for the delete_sso_tokens and has_sso_tokens functions.""" + from keyring.errors import KeyringError + api_host = "https://example.com" def test_delete_sso_tokens(self, mock_get_user, mock_delete_password): @@ -584,6 +600,8 @@ def test_delete_sso_tokens(self, mock_get_user, mock_delete_password): def test_delete_sso_tokens_handles_keyring_error( self, mock_get_user, mock_delete_password ): + from keyring.errors import KeyringError + mock_delete_password.side_effect = KeyringError("err") assert delete_sso_tokens(self.api_host) is False diff --git a/cloudsmith_cli/core/tests/test_metadata.py b/cloudsmith_cli/core/tests/test_metadata.py index b2f1e27d..01778a85 100644 --- a/cloudsmith_cli/core/tests/test_metadata.py +++ b/cloudsmith_cli/core/tests/test_metadata.py @@ -2,7 +2,6 @@ import json -import cloudsmith_api import httpretty import httpretty.core import pytest @@ -562,6 +561,8 @@ def test_401_when_unauthenticated(self): class TestAuthHeaders: @staticmethod def _override_config(monkeypatch, *, api_key=None, headers=None): + import cloudsmith_api + cfg = cloudsmith_api.Configuration() cfg.api_key = api_key if api_key is not None else cfg.api_key cfg.headers = headers if headers is not None else cfg.headers diff --git a/cloudsmith_cli/credential_helpers/cargo/__init__.py b/cloudsmith_cli/credential_helpers/cargo/__init__.py new file mode 100644 index 00000000..31a60cba --- /dev/null +++ b/cloudsmith_cli/credential_helpers/cargo/__init__.py @@ -0,0 +1,4 @@ +# Copyright 2026 Cloudsmith Ltd +from .runtime import execute, get_credentials, handle_request, hello + +__all__ = ["execute", "get_credentials", "handle_request", "hello"] diff --git a/cloudsmith_cli/credential_helpers/cargo/installer.py b/cloudsmith_cli/credential_helpers/cargo/installer.py new file mode 100644 index 00000000..18d7bc50 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/cargo/installer.py @@ -0,0 +1,491 @@ +# Copyright 2026 Cloudsmith Ltd +"""Installer for the Cargo credential provider. + +Manages writing/removing the ``cargo-credential-cloudsmith`` launcher and +patching ``$CARGO_HOME/config.toml`` so Cargo routes registry authentication +through the Cloudsmith credential provider. + +Cargo does not map credential providers to hostnames: a provider is registered +globally (``registry.global-credential-providers``) or per named registry +(``registries..credential-provider``), and is told the registry index URL +at call time. Installing therefore registers the +provider globally — safe because the runtime answers ``url-not-supported`` for +anything that is not a Cloudsmith registry, so Cargo falls through to the next +provider — and additionally pins it on any already-configured registry whose +index points at a known Cloudsmith Cargo host. +""" + +from __future__ import annotations + +import logging +import os +import sys +from pathlib import Path +from typing import TYPE_CHECKING + +import tomlkit + +from ...core.cache_utils import merge_config_file +from ..backends import BackendKind +from ..common import extract_hostname +from ..custom_domains import get_format_domains +from ..launchers import is_on_path, remove_launcher, resolve_bin_dir, write_launcher + +if TYPE_CHECKING: + from ...core.credentials.models import CredentialResult + +logger = logging.getLogger(__name__) + + +def _cargo_config_path() -> Path: + """Return the path to the Cargo configuration file. + + Respects the ``CARGO_HOME`` environment variable; otherwise returns the + platform default ``~/.cargo/config.toml``. Note this is *not* + ``credentials.toml``: credential providers are configured in + ``config.toml``, while ``credentials.toml`` holds tokens (which this + helper deliberately never writes — the token is resolved at call time). + """ + cargo_home = os.environ.get("CARGO_HOME") + base = Path(cargo_home) if cargo_home else Path.home() / ".cargo" + return base / "config.toml" + + +class CargoInstaller: + """Manages installation of the Cargo credential provider for Cloudsmith. + + This installer writes a ``cargo-credential-cloudsmith`` launcher binary and + patches ``$CARGO_HOME/config.toml`` to register it as a credential + provider. + + Usage:: + + installer = CargoInstaller() + actions = installer.install(domains=["my-registry.example.com"]) + for action in actions: + print(action) + """ + + LAUNCHER_NAME = "cargo-credential-cloudsmith" + TARGET_CMD = "cloudsmith credential-helper cargo" + #: The value written into Cargo's config. Cargo resolves a bare name + #: through ``PATH`` (and requires the ``cargo-credential-`` prefix). + PROVIDER_VALUE = "cargo-credential-cloudsmith" + #: Cargo's built-in token provider. Setting + #: ``global-credential-providers`` replaces Cargo's default of + #: ``["cargo:token"]``, so it has to be carried forward explicitly or + #: hand-written tokens in ``credentials.toml`` stop working. + TOKEN_PROVIDER = "cargo:token" + DEFAULT_HOST = "cargo.cloudsmith.io" + + name = "cargo" + summary = "Cargo credential provider for Cloudsmith registries" + + @classmethod + def _resolve_target_cmd(cls) -> str: + """Return the command the launcher forwards to. + + A pip/source install resolves the bare ``cloudsmith`` command via + ``PATH``. A frozen standalone binary (PyInstaller) is not guaranteed + to be on ``PATH`` under that name, so point the launcher at the + absolute executable instead — mirroring the frozen handling in + :func:`cloudsmith_cli.cli.commands.mcp._get_server_config`. The path + is quoted so a directory containing spaces still execs correctly. + """ + if getattr(sys, "frozen", False): + return f'"{sys.executable}" credential-helper cargo' + return cls.TARGET_CMD + + # ------------------------------------------------------------------ + # config.toml mutation + # ------------------------------------------------------------------ + + def _add_provider(self, config: dict, hosts: list[str]) -> None: + """Register the provider globally and on matching named registries.""" + registry = config.get("registry") + if not isinstance(registry, dict): + registry = config["registry"] = {} + + providers = registry.get("global-credential-providers") + if isinstance(providers, list): + providers = [p for p in providers if p != self.PROVIDER_VALUE] + else: + providers = [self.TOKEN_PROVIDER] + + # Cargo tries providers from last to first, so appending ours gives it + # the highest precedence: it answers for Cloudsmith registries and + # defers everything else back down the list. + providers.append(self.PROVIDER_VALUE) + registry["global-credential-providers"] = providers + + # A per-registry `credential-provider` shadows the global list, so pin + # ours on any registry whose index is a known Cloudsmith Cargo host. + for _name, entry in self._cloudsmith_registries(config, hosts): + entry["credential-provider"] = self.PROVIDER_VALUE + + def _remove_provider(self, config: dict) -> None: + """Strip every trace of this provider from a parsed config.""" + registry = config.get("registry") + if isinstance(registry, dict): + providers = registry.get("global-credential-providers") + if isinstance(providers, list): + remaining = [p for p in providers if p != self.PROVIDER_VALUE] + if remaining != providers: + # A list of just the built-in token provider is exactly + # Cargo's default, so drop the key and let the default + # apply rather than leaving our edit behind. + if remaining in ([], [self.TOKEN_PROVIDER]): + del registry["global-credential-providers"] + else: + registry["global-credential-providers"] = remaining + if not registry: + del config["registry"] + + registries = config.get("registries") + if isinstance(registries, dict): + for entry in registries.values(): + if ( + isinstance(entry, dict) + and entry.get("credential-provider") == self.PROVIDER_VALUE + ): + del entry["credential-provider"] + + @staticmethod + def _cloudsmith_registries( + config: dict, hosts: list[str] + ) -> list[tuple[str, dict]]: + """Return ``(name, entry)`` for ``[registries.*]`` whose index is in *hosts*.""" + registries = config.get("registries") + if not isinstance(registries, dict): + return [] + + wanted = {host.lower() for host in hosts} + matches: list[tuple[str, dict]] = [] + for name, entry in registries.items(): + if not isinstance(entry, dict): + continue + index = entry.get("index") + if isinstance(index, str) and extract_hostname(index) in wanted: + matches.append((name, entry)) + return matches + + # ------------------------------------------------------------------ + # install / uninstall / status + # ------------------------------------------------------------------ + + def install( + self, + *, + bin_dir: str | None = None, + domains: tuple[str, ...] = (), + discover: bool = True, + refresh: bool = False, + org: str | None = None, + credential: CredentialResult | None = None, + api_host: str | None = None, + dry_run: bool = False, + ) -> list[str]: + """Install the Cargo credential provider. + + Writes the launcher binary and registers it in + ``$CARGO_HOME/config.toml``. + + Parameters + ---------- + bin_dir: + Override for the directory to install the launcher. Defaults to + :func:`resolve_bin_dir` auto-detection. + domains: + Additional registry hostnames to recognise (in addition to the + default ``cargo.cloudsmith.io``) when pinning the provider on + named registries. + discover: + When ``True`` (default), attempt to auto-discover Cargo custom + domains via the Cloudsmith API. Discovery is best-effort and never + prevents the provider from being registered. + refresh: + When ``True``, bypass the domain cache and fetch fresh data from + the API. Only meaningful when *discover* is also ``True``. + org: + Cloudsmith organisation slug used for custom-domain discovery. + credential: + Resolved credential used for custom-domain discovery. + api_host: + Cloudsmith API host URL override. + dry_run: + When ``True``, compute and return planned actions without writing + any files. + + Returns + ------- + list[str] + Human-readable descriptions of actions taken (or planned, when + *dry_run* is ``True``). + """ + target_dir = resolve_bin_dir(bin_dir) + config_path = _cargo_config_path() + + actions: list[str] = [] + + # Start with the default host plus any explicitly requested domains. + hosts: list[str] = [self.DEFAULT_HOST, *domains] + + # --- Custom-domain auto-discovery (best-effort) --- + if discover: + if dry_run: + # Discovery queries the API and refreshes the on-disk domain + # cache, neither of which a "no changes" preview may do. + actions.append("skipped custom-domain auto-discovery (dry run)") + elif org and credential and credential.api_key: + # Discovery boundary: network/SDK errors must never abort the + # default install. ApiException is already handled inside + # get_format_domains; this broad catch is the deliberate outer + # boundary (consistent with "boundary catches, library stays clean"). + # Note: BaseException subclasses (KeyboardInterrupt/SystemExit) + # intentionally propagate — they are not caught by `except Exception`. + try: + discovered = get_format_domains( + org, + BackendKind.CARGO, + credential=credential, + api_host=api_host, + refresh=refresh, + ) + except Exception as exc: # pylint: disable=broad-except + # Discovery is best-effort: never let it abort the install of + # the defaults. (Network/SDK errors degrade to a warning; + # ApiException is already handled inside.) + actions.append( + f"WARNING: custom-domain auto-discovery failed: {exc}" + ) + discovered = [] + new_hosts = [h for h in discovered if h not in hosts] + hosts.extend(discovered) + actions.append( + f"discovered {len(new_hosts)} new Cargo custom domain(s)" + ) + else: + logger.debug( + "skipped auto-discovery" + " (no organization/credentials; pass --no-discover to silence)" + ) + + # De-duplicate while preserving order + seen: set[str] = set() + deduped: list[str] = [] + for h in hosts: + if h not in seen: + seen.add(h) + deduped.append(h) + hosts = deduped + + # Recorded for the action log: which named registries get pinned. + pinned: list[str] = [] + + def mutate(config: dict) -> None: + pinned.clear() + pinned.extend( + name for name, _ in self._cloudsmith_registries(config, hosts) + ) + self._add_provider(config, hosts) + + if dry_run: + if os.name == "nt": + launcher_path = target_dir / f"{self.LAUNCHER_NAME}.cmd" + else: + launcher_path = target_dir / self.LAUNCHER_NAME + actions.append(f"would write launcher {launcher_path}") + + would_change = merge_config_file( + config_path, mutate, dry_run=True, format="toml" + ) + if would_change: + actions.append( + f"would add {self.PROVIDER_VALUE!r} to" + f" registry.global-credential-providers in {config_path}" + ) + + actions.extend( + [ + ( + f"would set registries.{name}.credential-provider" + f"={self.PROVIDER_VALUE!r} in {config_path}" + ) + for name in pinned + ] + ) + else: + actions.append( + f"{self.PROVIDER_VALUE!r} already registered" + f" in {config_path} (no change)" + ) + return actions + + # Real install + launcher_path = write_launcher( + target_dir, self.LAUNCHER_NAME, self._resolve_target_cmd() + ) + actions.append(f"wrote launcher {launcher_path}") + + changed = merge_config_file(config_path, mutate, format="toml") + if changed: + actions.append( + f"added {self.PROVIDER_VALUE!r} to" + f" registry.global-credential-providers in {config_path}" + ) + actions.extend( + [ + ( + f"set registries.{name}.credential-provider" + f"={self.PROVIDER_VALUE!r} in {config_path}" + ) + for name in pinned + ] + ) + else: + actions.append(f"config.toml already up to date ({config_path})") + + if not is_on_path(target_dir): + actions.append( + f"WARNING: {target_dir} is not on PATH — " + f"add it to your PATH so Cargo can find {self.LAUNCHER_NAME}" + ) + + return actions + + def uninstall( + self, *, bin_dir: str | None = None, dry_run: bool = False + ) -> list[str]: + """Uninstall the Cargo credential provider. + + Removes the launcher binary and strips Cloudsmith-managed entries from + ``$CARGO_HOME/config.toml``. + + Parameters + ---------- + bin_dir: + Override for the directory where the launcher was installed. + Defaults to :func:`resolve_bin_dir` auto-detection. Pass the same + value that was given to :meth:`install` so the correct launcher file + is found and removed. + dry_run: + When ``True``, return planned actions without writing any files. + + Returns + ------- + list[str] + Human-readable descriptions of actions taken (or planned). + """ + target_dir = resolve_bin_dir(bin_dir) + config_path = _cargo_config_path() + + actions: list[str] = [] + + if os.name == "nt": + launcher_path = target_dir / f"{self.LAUNCHER_NAME}.cmd" + else: + launcher_path = target_dir / self.LAUNCHER_NAME + + # An absent config is nothing to strip: merging into it would create an + # empty config.toml as a parting gift. + config_exists = config_path.exists() + + if dry_run: + if launcher_path.exists(): + actions.append(f"would remove launcher {launcher_path}") + else: + actions.append( + f"launcher not found at {launcher_path} (nothing to remove)" + ) + + would_change = config_exists and merge_config_file( + config_path, self._remove_provider, dry_run=True, format="toml" + ) + if would_change: + actions.append( + f"would remove {self.PROVIDER_VALUE!r} entries from {config_path}" + ) + else: + actions.append( + f"no {self.PROVIDER_VALUE!r} entries to remove from {config_path}" + ) + return actions + + # Real uninstall + removed = remove_launcher(target_dir, self.LAUNCHER_NAME) + if removed: + actions.append(f"removed launcher {launcher_path}") + else: + actions.append(f"launcher not found at {launcher_path} (nothing to remove)") + + changed = config_exists and merge_config_file( + config_path, self._remove_provider, format="toml" + ) + if changed: + actions.append( + f"removed {self.PROVIDER_VALUE!r} entries from {config_path}" + ) + else: + actions.append( + f"no {self.PROVIDER_VALUE!r} entries to remove from {config_path}" + ) + + return actions + + def status(self) -> dict: + """Return current installation status. + + Returns + ------- + dict + A dict with keys: + + ``"launcher"`` + The path of the launcher if it exists, else ``None``. + ``"hosts"`` + Where the provider is registered in ``config.toml``: the index + hostnames of named registries pinned to it, plus a marker when + it is registered in ``registry.global-credential-providers`` + (which covers every registry, since Cargo passes the index URL + at call time rather than matching on hostname). + """ + target_dir = resolve_bin_dir() + if os.name == "nt": + launcher_path: Path | None = target_dir / f"{self.LAUNCHER_NAME}.cmd" + else: + launcher_path = target_dir / self.LAUNCHER_NAME + + if launcher_path is not None and not launcher_path.exists(): + launcher_path = None + + config_path = _cargo_config_path() + hosts: list[str] = [] + if config_path.exists(): + try: + data = tomlkit.loads(config_path.read_text(encoding="utf-8")) + except (tomlkit.exceptions.ParseError, ValueError, OSError): + data = {} + + registry = data.get("registry") + if isinstance(registry, dict): + providers = registry.get("global-credential-providers") + if isinstance(providers, list) and self.PROVIDER_VALUE in providers: + hosts.append("all registries (global-credential-providers)") + + registries = data.get("registries") + if isinstance(registries, dict): + for entry in registries.values(): + if ( + not isinstance(entry, dict) + or entry.get("credential-provider") != self.PROVIDER_VALUE + ): + continue + index = entry.get("index") + host = extract_hostname(index) if isinstance(index, str) else "" + if host and host not in hosts: + hosts.append(host) + + return { + "launcher": str(launcher_path) if launcher_path is not None else None, + "hosts": hosts, + } diff --git a/cloudsmith_cli/credential_helpers/cargo/runtime.py b/cloudsmith_cli/credential_helpers/cargo/runtime.py new file mode 100644 index 00000000..97e742bd --- /dev/null +++ b/cloudsmith_cli/credential_helpers/cargo/runtime.py @@ -0,0 +1,253 @@ +# Copyright 2026 Cloudsmith Ltd +""" +Cargo credential provider runtime. + +Transport-light protocol logic for the Cargo credential provider protocol. +This module is intentionally free of Click/sys imports so it can be unit-tested +without invoking the CLI machinery. + +Cargo speaks a line-delimited JSON *conversation* on stdin/stdout: the provider +announces the protocol versions it supports, then answers one request per line +until Cargo closes stdin. Every outcome is reported in-band as an +``{"Ok": ...}`` / ``{"Err": ...}`` message; a non-zero exit code is not how a +provider reports "no credential". + +See: https://doc.rust-lang.org/cargo/reference/credential-provider-protocol.html +""" + +import json +import logging + +from ..backends import BackendKind +from ..common import is_cloudsmith_domain + +logger = logging.getLogger(__name__) + +#: Protocol versions this provider implements, announced in the hello message. +SUPPORTED_VERSIONS = (1,) + +_REFUSAL_MESSAGE = ( + "Error: Unable to retrieve credentials. " + "Provide credentials via the CLOUDSMITH_API_KEY environment variable, " + "credentials.ini, the system keyring, or an OIDC service. " + "Verify current authentication with `cloudsmith whoami --verbose`." +) + + +def hello() -> dict: + """Return the handshake message sent before any request is read. + + Cargo waits for this line to learn which protocol versions the provider + speaks, and sets the ``v`` field of its requests to one of them. + """ + return {"v": list(SUPPORTED_VERSIONS)} + + +def _ok(payload: dict) -> dict: + """Wrap a success payload in the protocol's ``Ok`` envelope.""" + return {"Ok": payload} + + +def _err(kind: str, message: str | None = None) -> dict: + """Wrap an error *kind* in the protocol's ``Err`` envelope. + + ``url-not-supported`` tells Cargo to move on to the next configured + provider, so it is the correct answer for a non-Cloudsmith registry — + installing this helper globally must not break authentication to + crates.io or to a third-party registry. + """ + error: dict = {"kind": kind} + if message is not None: + error["message"] = message + return {"Err": error} + + +def is_supported_registry(index_url, credential=None, api_host=None, org=None) -> bool: + """Return True when *index_url* is a Cloudsmith Cargo registry. + + Args: + index_url: The registry index URL from the request (may carry Cargo's + ``sparse+`` prefix, which :func:`extract_hostname` strips) + credential: Pre-resolved CredentialResult, used to authenticate the + custom-domain lookup + api_host: Cloudsmith API host URL + org: Organisation slug whose custom domains to match against + """ + if not index_url: + return False + + return is_cloudsmith_domain( + index_url, + credential=credential, + api_host=api_host, + backend_kind=BackendKind.CARGO, + org=org, + ) + + +def get_credentials(index_url, credential=None, api_host=None, org=None): + """ + Get the token for a Cloudsmith Cargo registry. + + Verifies the index URL is a Cloudsmith registry (including custom domains) + and returns the token if one is available. + + Args: + index_url: The Cargo registry index URL + credential: Pre-resolved CredentialResult from the provider chain + api_host: Cloudsmith API host URL + org: Organisation slug whose custom domains to match against + + Returns: + str: The token Cargo sends as its ``Authorization`` header, or None + """ + if not credential or not credential.api_key: + return None + + if not is_supported_registry( + index_url, credential=credential, api_host=api_host, org=org + ): + return None + + return credential.api_key + + +def _handle_get(request, credential, api_host, org) -> dict: + """Answer a ``get`` request with a token, or with why there isn't one.""" + registry = request.get("registry") + if not isinstance(registry, dict): + return _err("other", "Request is missing registry information") + + index_url = registry.get("index-url") + if not index_url: + return _err("other", "Request is missing the registry index-url") + + # Order matters: a registry we don't serve must fall through to the next + # provider (url-not-supported) whether or not we hold a credential, and a + # Cloudsmith registry we can't authenticate is not-found rather than + # unsupported. + if not is_supported_registry( + index_url, credential=credential, api_host=api_host, org=org + ): + return _err("url-not-supported") + + if not credential or not credential.api_key: + return _err("not-found") + + return _ok( + { + "kind": "get", + "token": credential.api_key, + # A Cloudsmith token is organisation-wide and not scoped to the + # read/publish/yank/owners operation, so it is cacheable for the + # session and across operations. + "cache": "session", + "operation_independent": True, + } + ) + + +def handle_request(request, credential=None, api_host=None, org=None) -> dict: + """ + Answer a single decoded Cargo credential-provider request. + + Args: + request: The decoded request object (one JSON line from Cargo) + credential: Pre-resolved CredentialResult from the provider chain + api_host: Cloudsmith API host URL + org: Organisation slug whose custom domains to match against + + Returns: + dict: The response message to serialise back to Cargo. + """ + if not isinstance(request, dict): + return _err("other", "Request is not a JSON object") + + version = request.get("v") + if version not in SUPPORTED_VERSIONS: + return _err( + "other", + f"Unsupported protocol version {version!r}" + f" (supported: {', '.join(str(v) for v in SUPPORTED_VERSIONS)})", + ) + + kind = request.get("kind") + if kind == "get": + return _handle_get(request, credential, api_host, org) + + if kind in ("login", "logout"): + # Credentials are resolved from the Cloudsmith CLI's own provider chain + # (API key, credentials.ini, keyring, OIDC), so there is nothing for + # `cargo login`/`cargo logout` to store or clear here. + return _err("operation-not-supported") + + return _err("operation-not-supported") + + +def _write_message(stdout, message) -> None: + """Write one newline-delimited JSON message and flush it. + + Cargo reads line by line and blocks on the next message, so an unflushed + buffer deadlocks the exchange. + """ + stdout.write(json.dumps(message) + "\n") + stdout.flush() + + +def execute(stdin, stdout, credential=None, api_host=None, org=None): + """ + Run a Cargo credential-provider session. + + Emits the hello message, then answers one request per line of *stdin* + until Cargo closes it. + + Args: + stdin: A file-like object to read newline-delimited requests from + stdout: A file-like object to write newline-delimited responses to + credential: Pre-resolved CredentialResult from the provider chain + api_host: Cloudsmith API host URL + org: Organisation slug whose custom domains to match against + + Returns: + A (exit_code, stderr_text) tuple. Protocol-level outcomes are reported + in-band to Cargo, so the exit code only distinguishes a clean session + from one that could not answer for lack of credentials (or that broke + at the transport level); *stderr_text* is None when there is nothing to + tell the user. + """ + refused = False + + try: + _write_message(stdout, hello()) + + for line in stdin: + line = line.strip() + if not line: + continue + + try: + request = json.loads(line) + except (json.JSONDecodeError, ValueError) as exc: + response = _err("other", f"Malformed request: {exc}") + else: + response = handle_request( + request, credential=credential, api_host=api_host, org=org + ) + + if response.get("Err", {}).get("kind") == "not-found": + refused = True + + _write_message(stdout, response) + except Exception as exc: # pylint: disable=broad-except + # Protocol boundary: a credential provider must never crash `cargo + # build`/`publish` with a traceback. Covers broken-pipe OSError from + # stdin/stdout, network/SDK errors from the custom-domain lookup, and + # TypeError from json.dumps — all degrade to a clean exit. + # (Exception does not catch KeyboardInterrupt/SystemExit, which is correct.) + logger.debug("cargo credential-provider session failed: %s", exc, exc_info=True) + return (1, _REFUSAL_MESSAGE) + + if refused: + return (1, _REFUSAL_MESSAGE) + + return (0, None) diff --git a/cloudsmith_cli/credential_helpers/docker/installer.py b/cloudsmith_cli/credential_helpers/docker/installer.py index cf25b744..e4aa4df7 100644 --- a/cloudsmith_cli/credential_helpers/docker/installer.py +++ b/cloudsmith_cli/credential_helpers/docker/installer.py @@ -15,7 +15,7 @@ from pathlib import Path from typing import TYPE_CHECKING -from ...core.cache_utils import merge_json_file +from ...core.cache_utils import merge_config_file from ..backends import BackendKind from ..custom_domains import get_format_domains from ..launchers import is_on_path, remove_launcher, resolve_bin_dir, write_launcher @@ -195,7 +195,7 @@ def mutate(config: dict) -> None: launcher_path = target_dir / self.LAUNCHER_NAME actions.append(f"would write launcher {launcher_path}") - would_change = merge_json_file(config_path, mutate, dry_run=True) + would_change = merge_config_file(config_path, mutate, dry_run=True) for host in hosts: if would_change: actions.append( @@ -215,7 +215,7 @@ def mutate(config: dict) -> None: ) actions.append(f"wrote launcher {launcher_path}") - changed = merge_json_file(config_path, mutate) + changed = merge_config_file(config_path, mutate) if changed: actions.extend( f"set credHelpers[{host!r}]={self.HELPER_VALUE!r} in {config_path}" @@ -283,7 +283,7 @@ def mutate(config: dict) -> None: f"launcher not found at {launcher_path} (nothing to remove)" ) - would_change = merge_json_file(config_path, mutate, dry_run=True) + would_change = merge_config_file(config_path, mutate, dry_run=True) if would_change: actions.append( f"would remove credHelpers entries with value" @@ -300,7 +300,7 @@ def mutate(config: dict) -> None: else: actions.append(f"launcher not found at {launcher_path} (nothing to remove)") - changed = merge_json_file(config_path, mutate) + changed = merge_config_file(config_path, mutate) if changed: actions.append( f"removed credHelpers entries with value" diff --git a/pyproject.toml b/pyproject.toml index 9d7b69c9..f65d1bcf 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -58,6 +58,7 @@ dependencies = [ "rich>=13.0.0", "semver>=2.7.9", "urllib3>=2.5", + "tomlkit>=0.15.0", ] [project.optional-dependencies] diff --git a/uv.lock b/uv.lock index 1f6e67bc..9f5c6e66 100644 --- a/uv.lock +++ b/uv.lock @@ -518,6 +518,7 @@ dependencies = [ sdist = { url = "https://files.pythonhosted.org/packages/3e/7c/c0e827e54989dc5031b95ceb6727997ea97b3bb5f22b9482dafa282f6798/cloudsmith_api-2.0.31.tar.gz", hash = "sha256:768888677ca6570dde5ea1b9b3254401786a1e4c2842c170ce715e1796306c91", size = 727667, upload-time = "2026-08-20T12:37:10.543Z" } wheels = [ { url = "https://files.pythonhosted.org/packages/6b/15/a5bcf3f2baeb403acb3dde53a7175bb9c06fc6d9844d324bf27f9f913cca/cloudsmith_api-2.0.31-py2.py3-none-any.whl", hash = "sha256:f6d8bd2889e5b58b06a81f2980911bae589631161fd2b4524cae346de0d72d4e", size = 1394274, upload-time = "2026-08-20T12:37:09.237Z" }, + { url = "https://files.pythonhosted.org/packages/76/84/e05f7fa12ed1a69b4877d9d77733864db746d33f356a8fcb92a3203e92a2/cloudsmith_api-2.0.31-py3-none-any.whl", hash = "sha256:958b0971de2a06d312bac71fece99e98ce9610edd95e2141304d75faa146a431", size = 1178628, upload-time = "2026-08-25T10:36:13.481Z" }, ] [[package]] @@ -540,6 +541,7 @@ dependencies = [ { name = "requests-toolbelt" }, { name = "rich" }, { name = "semver" }, + { name = "tomlkit" }, { name = "urllib3" }, ] @@ -596,6 +598,7 @@ requires-dist = [ { name = "requests-toolbelt", specifier = ">=1.0.0" }, { name = "rich", specifier = ">=13.0.0" }, { name = "semver", specifier = ">=2.7.9" }, + { name = "tomlkit", specifier = ">=0.15.0" }, { name = "urllib3", specifier = ">=2.5" }, ] provides-extras = ["aws", "all"]