diff --git a/CHANGELOG.md b/CHANGELOG.md index ad63325c..70832bb1 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added +- Added a NuGet credential provider. Run `cloudsmith credential-helper install nuget --workspace ` once. Then `dotnet restore`, MSBuild and Visual Studio authenticate to Cloudsmith feeds with your CLI credentials. NuGet finds the provider on PATH. This needs .NET SDK 9.0.200 or later. The provider also covers the NuGet custom domains of the Workspace. To name feed hosts yourself, use `--domain`. To skip the domain lookup, use `--no-discover`. No token is stored in `nuget.config`. - Added `skills` to `cloudsmith upstream`. Use `cloudsmith upstream skills list`, `create`, `update`, and `delete` to manage AI Skills upstreams. An AI Skills upstream caches skills and plugins from a GitHub repository. - Added `--changelog` to `cloudsmith push skills`. Use it to add release notes to an AI Skills package. The option is available only for the `skills` format. - Added `--compress` to `cloudsmith push skills`. With this option, you can pass a skill folder instead of a `.zip` archive. The CLI compresses the contents of the folder into a `.zip` archive, then uploads it. If you pass a folder without `--compress`, the command fails with a usage error. The option is available only for the `skills` format. diff --git a/cloudsmith_cli/cli/commands/credential_helper/__init__.py b/cloudsmith_cli/cli/commands/credential_helper/__init__.py index bcf76528..412fd3c2 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/__init__.py +++ b/cloudsmith_cli/cli/commands/credential_helper/__init__.py @@ -13,6 +13,7 @@ from .docker import docker as docker_cmd from .generic import generic as generic_cmd from .manage import install_cmd, list_cmd, uninstall_cmd +from .nuget import nuget as nuget_cmd from .pnpm import pnpm as pnpm_cmd from .terraform import terraform as terraform_cmd @@ -37,6 +38,9 @@ def credential_helper(): \b # Install cargo credential helper $ cloudsmith credential-helper install cargo + \b + # Install NuGet credential provider + $ cloudsmith credential-helper install nuget \b # Test Docker credential helper directly $ echo "docker.cloudsmith.io" | cloudsmith credential-helper docker @@ -54,5 +58,6 @@ def credential_helper(): credential_helper.add_command(list_cmd, name="list") credential_helper.add_command(cargo_cmd, name="cargo") credential_helper.add_command(terraform_cmd, name="terraform") +credential_helper.add_command(nuget_cmd, name="nuget") main.add_command(credential_helper, name="credential-helper") diff --git a/cloudsmith_cli/cli/commands/credential_helper/manage.py b/cloudsmith_cli/cli/commands/credential_helper/manage.py index 34475753..c709bbdf 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/manage.py +++ b/cloudsmith_cli/cli/commands/credential_helper/manage.py @@ -18,6 +18,7 @@ from cloudsmith_cli.credential_helpers.cargo.installer import CargoInstaller from cloudsmith_cli.credential_helpers.generic import PartialInstallError from cloudsmith_cli.credential_helpers.launchers import UnsafeLauncherValue +from cloudsmith_cli.credential_helpers.nuget.installer import NuGetInstaller from cloudsmith_cli.credential_helpers.pnpm.installer import PNPMInstaller from cloudsmith_cli.credential_helpers.terraform.installer import ( TerraformHelperExeNotFound, @@ -47,6 +48,7 @@ "pnpm": PNPMInstaller, "cargo": CargoInstaller, "terraform": TerraformInstaller, + "nuget": NuGetInstaller, } diff --git a/cloudsmith_cli/cli/commands/credential_helper/nuget.py b/cloudsmith_cli/cli/commands/credential_helper/nuget.py new file mode 100644 index 00000000..e84637ca --- /dev/null +++ b/cloudsmith_cli/cli/commands/credential_helper/nuget.py @@ -0,0 +1,104 @@ +# Copyright 2026 Cloudsmith Ltd +""" +NuGet credential provider command. + +Implements the NuGet cross-platform authentication plugin protocol for +Cloudsmith feeds. + +See: + https://learn.microsoft.com/en-us/nuget/reference/extensibility/nuget-cross-platform-plugins +""" + +import sys + +import click + +from ....core import keyring +from ....core.credentials.chain import CredentialProviderChain +from ....credential_helpers.nuget import execute +from ...decorators import common_api_auth_options, initialise_api + + +def _utf8(stream, **kwargs): + """Switch a text stream to UTF-8, which the NuGet plugin protocol requires.""" + try: + stream.reconfigure(encoding="utf-8", **kwargs) + except (AttributeError, ValueError, OSError): + pass + return stream + + +def _resolve_credential(opts, non_interactive): + """Resolve the credential and keep it on *opts*, where telemetry redacts it.""" + if non_interactive: + keyring.forbid_password_prompts() + opts.credential = CredentialProviderChain().resolve(opts.credential_context) + return opts.credential + + +@click.command( + context_settings={"ignore_unknown_options": True, "allow_interspersed_args": False} +) +@click.option( + "--domain", + "domains", + multiple=True, + help="A feed host to authenticate, as well as the Workspace's NuGet domains.", +) +@click.option( + "--no-discover", + is_flag=True, + help="Do not look up the Workspace's NuGet domains.", +) +@click.argument("nuget_args", nargs=-1, type=click.UNPROCESSED) +@common_api_auth_options +# NuGet gives the handshake a deadline, so a slow credential, such as an OIDC +# exchange, resolves at the first credentials request. +@initialise_api(defer_credentials=True) +def nuget(opts, domains, no_discover, nuget_args): + """ + NuGet credential provider for Cloudsmith feeds. + + NuGet (dotnet, MSBuild, Visual Studio) runs this command through the + ``nuget-plugin-cloudsmith`` launcher that ``cloudsmith credential-helper + install nuget`` puts on PATH, passing ``-Plugin`` after ``--``. It then + speaks the NuGet cross-platform plugin protocol on stdin/stdout. + + Provides credentials for all Cloudsmith NuGet feeds: ``*.cloudsmith.io``, + ``*.cloudsmith.com``, and any NuGet custom domains configured for the + Workspace (requires a Workspace - ``--workspace``, CLOUDSMITH_WORKSPACE, + or ``workspace`` in ``config.ini``; legacy aliases are also accepted - and + a valid API key/token). Any other feed is declined so NuGet falls back to + its other credential sources. + + \b + Examples: + + \b + # Called by NuGet via the launcher + $ nuget-plugin-cloudsmith -Plugin + + \b + Environment variables: + CLOUDSMITH_API_KEY: API key for authentication (optional) + CLOUDSMITH_WORKSPACE: Workspace slug (CLOUDSMITH_ORG is also accepted) + """ + protocol_stdout = _utf8(sys.stdout, newline="\n") + # Other output, such as the HTTP trace of --debug, must not reach NuGet. + sys.stdout = sys.stderr + exit_code, stderr = execute( + nuget_args, + _utf8(sys.stdin, errors="replace"), + protocol_stdout, + api_host=opts.api_host, + workspace=opts.org, + refresh_credential=lambda non_interactive: _resolve_credential( + opts, non_interactive + ), + domains=domains, + discover=not no_discover, + ) + + if stderr is not None: + click.echo(stderr, err=True) + sys.exit(exit_code) diff --git a/cloudsmith_cli/cli/decorators.py b/cloudsmith_cli/cli/decorators.py index 8f790458..d0e1abf4 100644 --- a/cloudsmith_cli/cli/decorators.py +++ b/cloudsmith_cli/cli/decorators.py @@ -697,8 +697,13 @@ def _warn_on_oidc_detector_controls(order, disabled): ) -def resolve_credentials(f): - """Resolve credentials via the provider chain. Depends on initialise_session.""" +def resolve_credentials(f=None, *, defer=False): + """Resolve credentials via the provider chain. Depends on initialise_session. + + With ``defer=True``, the command resolves them from ``opts.credential_context``. + """ + if f is None: + return functools.partial(resolve_credentials, defer=defer) @click.option( "--oidc-audience", @@ -784,7 +789,7 @@ def wrapper(ctx, *args, **kwargs): ) chain = CredentialProviderChain() - credential = chain.resolve(context) + credential = None if defer else chain.resolve(context) if context.keyring_refresh_failed and not is_auth_command: if credential: @@ -812,6 +817,7 @@ def wrapper(ctx, *args, **kwargs): click.secho(message, fg="yellow", err=True) opts.credential = credential + opts.credential_context = context kwargs["opts"] = opts return ctx.invoke(f, *args, **kwargs) @@ -819,8 +825,10 @@ def wrapper(ctx, *args, **kwargs): return initialise_session(wrapper) -def initialise_api(f): +def initialise_api(f=None, *, defer_credentials=False): """Initialise the Cloudsmith API for use. Depends on resolve_credentials.""" + if f is None: + return functools.partial(initialise_api, defer_credentials=defer_credentials) @click.option( "-R", @@ -897,7 +905,7 @@ def call_print_rate_limit_info_with_opts(rate_info): kwargs["opts"] = opts return ctx.invoke(f, *args, **kwargs) - return resolve_credentials(wrapper) + return resolve_credentials(wrapper, defer=defer_credentials) def initialise_mcp(f): 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 41fcca7d..70836012 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py @@ -24,8 +24,8 @@ from ....credential_helpers.launchers import ( UnsafeLauncherValue, _launcher_content, - _launcher_filename, is_on_path, + launcher_filename, remove_launcher, resolve_bin_dir, write_launcher, @@ -103,7 +103,7 @@ def test_launcher_filename_and_content( ``NotImplementedError`` on Python < 3.12). """ assert ( - _launcher_filename(f"{format}-credential-cloudsmith", windows=windows) + launcher_filename(f"{format}-credential-cloudsmith", windows=windows) == expected_name ) assert ( diff --git a/cloudsmith_cli/cli/tests/test_credential_helper_nuget.py b/cloudsmith_cli/cli/tests/test_credential_helper_nuget.py new file mode 100644 index 00000000..b95d0037 --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_credential_helper_nuget.py @@ -0,0 +1,729 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the `cloudsmith credential-helper nuget` command and installer.""" + +from __future__ import annotations + +import collections +import io +import json +import os +import sys +import threading +from pathlib import Path +from unittest.mock import Mock, patch + +import click.testing +import cloudsmith_api +import pytest + +from ...core import keyring as core_keyring +from ...core.api.exceptions import ApiException +from ...core.credentials.models import CredentialResult +from ...credential_helpers.common import REFUSAL_MESSAGE +from ...credential_helpers.nuget.installer import NuGetInstaller +from ...credential_helpers.nuget.runtime import ( + PluginSession, + is_plugin_mode, + is_supported_source, + negotiate_version, + parse_timespan, +) +from ..commands.credential_helper.manage import install_cmd, list_cmd +from ..commands.credential_helper.nuget import nuget +from ..commands.main import main +from ..config import OPTIONS + +CLOUDSMITH_FEED = "https://nuget.cloudsmith.io/acme/repo/v3/index.json" +CUSTOM_FEED = "https://nuget.acme.com/repo/v3/index.json" +NUGET_ORG = "https://api.nuget.org/v3/index.json" +FORMAT_DOMAINS = "cloudsmith_cli.credential_helpers.common.get_format_domains" +LAUNCHER = "nuget-plugin-cloudsmith" + + +@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 _message(request_id, method, payload=None, message_type="Request") -> dict: + message = {"RequestId": request_id, "Type": message_type, "Method": method} + if payload is not None: + message["Payload"] = payload + return message + + +HANDSHAKE = _message( + "h", "Handshake", {"ProtocolVersion": "2.0.0", "MinimumProtocolVersion": "1.0.0"} +) + + +def _credentials_request(uri=CLOUDSMITH_FEED) -> dict: + return _message( + "c", + "GetAuthenticationCredentials", + {"Uri": uri, "IsRetry": False, "IsNonInteractive": True}, + ) + + +class _NuGetFeed: + """NuGet's end of stdin: it sends Close or ends the input only after each answer. + + A message can be a raw line, which needs no answer. + """ + + def __init__(self, messages): + self.messages = messages + self.requests = collections.Counter() + self.answers = collections.Counter() + self.answered = threading.Condition() + + def record(self, send): + """Wrap ``PluginSession._send`` so that the feed sees each answer.""" + + def send_and_record(session, request_id, message_type, method, payload=None): + send(session, request_id, message_type, method, payload) + if message_type not in ("Request", "Progress"): + with self.answered: + self.answers[request_id] += 1 + self.answered.notify_all() + + return send_and_record + + def _wait_for_answers(self): + with self.answered: + self.answered.wait_for(lambda: self.answers >= self.requests, timeout=2) + + def lines(self): + """Yield each message as the line that NuGet writes.""" + for message in self.messages: + if isinstance(message, str): + yield message + "\n" + continue + if message.get("Method") == "Close": + self._wait_for_answers() + elif message.get("Type") == "Request": + self.requests[message["RequestId"]] += 1 + yield json.dumps(message, ensure_ascii=False) + "\n" + self._wait_for_answers() + + +def _session(*messages, credential=None, workspace=None, stdout=None): + """Run a plugin session over *messages*; return (code, stderr, responses).""" + feed = _NuGetFeed(messages) + stdout = io.StringIO() if stdout is None else stdout + with patch.object(PluginSession, "_send", feed.record(PluginSession._send)): + code, stderr = PluginSession( + feed.lines(), stdout, credential=credential, workspace=workspace + ).run() + sent = [json.loads(line) for line in stdout.getvalue().splitlines()] + return code, stderr, sent + + +def _run_cli(runner, args, *messages, command=nuget, encoding="utf-8") -> list[dict]: + """Run the CLI over *messages* and return the messages on stdout.""" + feed = _NuGetFeed(messages) + read_fd, write_fd = os.pipe() + + def write_feed(): + with open(write_fd, "w", encoding=encoding) as pipe: + for line in feed.lines(): + pipe.write(line) + pipe.flush() + + threading.Thread(target=write_feed, daemon=True).start() + with ( + open(read_fd, "rb") as stdin, + patch.object(PluginSession, "_send", feed.record(PluginSession._send)), + ): + result = runner.invoke(command, args=args, input=stdin, catch_exceptions=False) + return [json.loads(line) for line in result.stdout.splitlines()] + + +def _responses(sent) -> dict: + """Map RequestId -> message for everything the plugin answered.""" + return {m["RequestId"]: m for m in sent if m["Type"] != "Request"} + + +@pytest.mark.parametrize( + "args,expected", + [(["-Plugin"], True), (["/plugin"], True), (["-Uri", CLOUDSMITH_FEED], False)], +) +def test_is_plugin_mode(args, expected): + assert is_plugin_mode(args) is expected + + +@pytest.mark.parametrize( + "uri,workspace,expected", + [ + (CLOUDSMITH_FEED, None, True), + (NUGET_ORG, "acme", False), + (CUSTOM_FEED, None, False), + (CUSTOM_FEED, "acme", True), + ("http://nuget.cloudsmith.io/acme/repo/v3/index.json", None, False), + ("https://attacker.example?x=.cloudsmith.io", None, False), + (123, None, False), + ("https://attacker.example\\@nuget.cloudsmith.io/v3/index.json", None, False), + ], +) +def test_is_supported_source(credential, uri, workspace, expected): + """Only https Cloudsmith feeds, and NuGet domains of a Workspace, match.""" + with patch(FORMAT_DOMAINS, return_value=["nuget.acme.com"]): + assert ( + is_supported_source(uri, credential=credential, workspace=workspace) + is expected + ) + + +@pytest.mark.parametrize( + "uri,expected", + [ + (CUSTOM_FEED, True), + (CLOUDSMITH_FEED, True), + ("https://nuget.other.com/v3/index.json", False), + ], +) +def test_is_supported_source_without_discovery(credential, uri, expected): + """Explicit domains and Cloudsmith hosts match with no API lookup.""" + with patch(FORMAT_DOMAINS, side_effect=AssertionError("looked up")): + assert ( + is_supported_source( + uri, + credential=credential, + workspace="acme", + domains={"nuget.acme.com"}, + discover=False, + ) + is expected + ) + + +@pytest.mark.parametrize( + "payload,expected", + [ + ({"ProtocolVersion": "3.0.0", "MinimumProtocolVersion": "1.0.0"}, "2.0.0"), + ({"ProtocolVersion": "1.0.0", "MinimumProtocolVersion": "1.0.0"}, None), + ({"ProtocolVersion": "x"}, None), + ], +) +def test_negotiate_version(payload, expected): + assert negotiate_version(payload) == expected + + +def test_plugin_full_conversation(credential): + code, stderr, sent = _session( + HANDSHAKE, + _message("i", "Initialize", {"RequestTimeout": "00:00:05"}), + _message("o", "GetOperationClaims", {}), + _credentials_request(), + _message("x", "Close"), + credential=credential, + ) + responses = _responses(sent) + + assert (code, stderr) == (0, None) + assert responses["h"]["Payload"] == { + "ResponseCode": "Success", + "ProtocolVersion": "2.0.0", + } + assert responses["i"]["Payload"] == {"ResponseCode": "Success"} + assert responses["o"]["Payload"] == {"Claims": ["Authentication"]} + assert responses["c"]["Payload"] == { + "ResponseCode": "Success", + "Username": "token", + "Password": "k_abc", + "AuthenticationTypes": ["Basic"], + } + assert "x" not in responses + + +def test_plugin_claims_no_download_operations(credential): + claims = _message( + "o", + "GetOperationClaims", + {"PackageSourceRepository": CLOUDSMITH_FEED, "ServiceIndex": "{}"}, + ) + _, _, sent = _session(claims, credential=credential) + assert _responses(sent)["o"]["Payload"] == {"Claims": []} + + +def test_plugin_lookup_failure_is_not_found(credential, tmp_path): + """A failed domain lookup declines the feed, so NuGet tries other sources.""" + with ( + patch( + "cloudsmith_cli.credential_helpers.domains.get_cache_path", + return_value=tmp_path / "acme.json", + ), + patch( + "cloudsmith_cli.core.api.workspaces.list_domains", + side_effect=ApiException(0, detail="Connection refused"), + ), + ): + code, _, sent = _session( + _credentials_request(CUSTOM_FEED), credential=credential, workspace="acme" + ) + assert code == 0 + assert _responses(sent)["c"]["Payload"]["ResponseCode"] == "NotFound" + + +def test_plugin_reports_missing_credentials_for_a_cloudsmith_feed(): + code, stderr, sent = _session(_credentials_request()) + + payload = _responses(sent)["c"]["Payload"] + assert payload == {"ResponseCode": "Error", "Message": REFUSAL_MESSAGE} + assert (code, stderr) == (1, REFUSAL_MESSAGE) + + +@pytest.mark.parametrize( + "first_line,answered", + [ + ("not json", {"c": "Response"}), + (json.dumps(_message("u", "Unknown")), {"u": "Fault", "c": "Response"}), + ( + json.dumps( + _message("ours", "Handshake", {"ResponseCode": "Error"}, "Response") + ), + {}, + ), + ], + ids=["malformed-line", "unknown-method", "handshake-rejected"], +) +def test_plugin_session_handles_unexpected_messages(credential, first_line, answered): + with patch( + "cloudsmith_cli.credential_helpers.nuget.runtime.uuid.uuid4", + return_value="ours", + ): + _, _, sent = _session(first_line, _credentials_request(), credential=credential) + + assert {m["RequestId"]: m["Type"] for m in sent[1:]} == answered + + +class _MessageWatcher(io.StringIO): + """A stdout that sets ``seen`` when the plugin sends a message of one type.""" + + def __init__(self, message_type): + super().__init__() + self.message_type = message_type + self.seen = threading.Event() + + def write(self, text): + if json.loads(text)["Type"] == self.message_type: + self.seen.set() + return super().write(text) + + +def test_plugin_sends_progress_while_a_request_runs(credential): + """Progress before each Initialize deadline keeps NuGet waiting.""" + stdout = _MessageWatcher("Progress") + initialize = _message("i", "Initialize", {"RequestTimeout": "00:00:00.0200000"}) + with patch( + "cloudsmith_cli.credential_helpers.nuget.runtime.is_cloudsmith_domain", + lambda *args, **kwargs: stdout.seen.wait(timeout=1), + ): + _, _, sent = _session( + initialize, + _credentials_request(CUSTOM_FEED), + credential=credential, + workspace="acme", + stdout=stdout, + ) + + answers = [m for m in sent if m["RequestId"] == "c"] + assert answers[0] == { + "RequestId": "c", + "Type": "Progress", + "Method": "GetAuthenticationCredentials", + "Payload": {}, + } + assert {m["Type"] for m in answers[:-1]} == {"Progress"} + assert answers[-1]["Type"] == "Response" + assert answers[-1]["Payload"]["ResponseCode"] == "Success" + + +def test_plugin_answers_a_cancel_while_a_request_runs(credential): + """A Cancel gets a Cancel at once, and the late result is not sent.""" + stdout = _MessageWatcher("Cancel") + cancel = _message("c", "GetAuthenticationCredentials", message_type="Cancel") + with patch( + "cloudsmith_cli.credential_helpers.nuget.runtime.is_cloudsmith_domain", + lambda *args, **kwargs: stdout.seen.wait(timeout=1), + ): + _, _, sent = _session( + _credentials_request(CUSTOM_FEED), + cancel, + credential=credential, + workspace="acme", + stdout=stdout, + ) + + assert [(m["RequestId"], m["Type"]) for m in sent[1:]] == [("c", "Cancel")] + + +def _lines_then(lines, error): + """Yield *lines* as NuGet's stdin, then raise *error* if it is given.""" + yield from lines + if error is not None: + raise error + + +@pytest.mark.parametrize( + "after,error,code", + [ + ([json.dumps(_message("x", "Close"))], None, 0), + ([], None, 0), + ([], OSError("stdin closed"), 1), + ], + ids=["close", "end-of-input", "stdin-error"], +) +def test_plugin_ends_the_session_while_a_request_runs(credential, after, error, code): + """NuGet has gone, so the plugin drops the late answer and does not wait for it.""" + entered, release = threading.Event(), threading.Event() + daemon = [] + + def slow_lookup(*args, **kwargs): + daemon.append(threading.current_thread().daemon) + entered.set() + return release.wait(timeout=1) + + lines = [json.dumps(_credentials_request(CUSTOM_FEED)), *after] + stdout = io.StringIO() + with patch( + "cloudsmith_cli.credential_helpers.nuget.runtime.is_cloudsmith_domain", + slow_lookup, + ): + result = PluginSession( + _lines_then(lines, error), stdout, credential=credential, workspace="acme" + ).run() + assert entered.wait(timeout=1) + release.set() + + sent = [json.loads(line) for line in stdout.getvalue().splitlines()] + assert result == (code, None) + assert [m["Method"] for m in sent] == ["Handshake"] + assert daemon == [True] + + +@pytest.mark.filterwarnings("ignore::pytest.PytestUnhandledThreadExceptionWarning") +def test_plugin_resolves_again_after_a_failed_resolution(credential): + """A first failure does not stay, so a later request can get the credential.""" + resolve = Mock(side_effect=[None, credential]) + session = PluginSession(io.StringIO(), io.StringIO(), refresh_credential=resolve) + payload = _credentials_request()["Payload"] + + first = session.handle_request("GetAuthenticationCredentials", payload) + second = session.handle_request("GetAuthenticationCredentials", payload) + + assert first["ResponseCode"] == "Error" + assert second["Password"] == "k_abc" + + +def test_plugin_without_discovery_declines_a_feed_before_it_resolves(credential): + """The feed of another provider does not wait for a credential.""" + resolve = Mock(return_value=credential) + session = PluginSession( + io.StringIO(), io.StringIO(), refresh_credential=resolve, discover=False + ) + payload = _credentials_request(CUSTOM_FEED)["Payload"] + + response = session.handle_request("GetAuthenticationCredentials", payload) + + assert response["ResponseCode"] == "NotFound" + assert not resolve.called + + +def test_plugin_faults_a_request_that_fails(credential): + """A request that raises gets a Fault, so NuGet does not wait for it.""" + with patch( + "cloudsmith_cli.credential_helpers.nuget.runtime.is_cloudsmith_domain", + side_effect=RuntimeError("lookup failed"), + ): + _, _, sent = _session( + _credentials_request(CUSTOM_FEED), credential=credential, workspace="acme" + ) + + assert _responses(sent)["c"]["Type"] == "Fault" + + +@pytest.mark.parametrize( + "value,expected", + [("00:00:05", 5.0), ("1.02:03:04.5", 93784.5), ("5", None), (None, None)], +) +def test_parse_timespan(value, expected): + assert parse_timespan(value) == expected + + +class _BrokenPipe(io.StringIO): + def write(self, _): + raise BrokenPipeError("gone") + + +def test_plugin_transport_failure_exits_without_a_message(credential): + """NuGet can close the pipes before the handshake; that is not a credential error.""" + code, stderr = PluginSession( + io.StringIO(), _BrokenPipe(), credential=credential + ).run() + assert (code, stderr) == (1, None) + + +def test_cli_speaks_the_plugin_protocol(runner): + sent = _run_cli( + runner, ["-k", "k_abc", "-Plugin"], HANDSHAKE, _credentials_request() + ) + + assert sent[0]["Method"] == "Handshake" + assert _responses(sent)["c"]["Payload"]["Password"] == "k_abc" + + +def test_cli_reads_input_that_is_not_utf8(runner): + """On Windows, NuGet writes to the plugin in the console code page.""" + proxy_login = _message("p", "SetCredentials", {"ProxyPassword": "\u00e9t\u00e9"}) + sent = _run_cli( + runner, + ["-k", "k_abc", "-Plugin"], + proxy_login, + _credentials_request(), + encoding="cp1252", + ) + + assert _responses(sent)["c"]["Payload"]["Password"] == "k_abc" + + +def test_cli_looks_up_domains_with_the_proxy_and_tls_settings(runner, tmp_path): + """A domain-cache miss reaches the API through the configured proxy.""" + seen = {} + + def list_domains(workspace, credential=None): + config = cloudsmith_api.Configuration() + seen.update(proxy=config.proxy, verify_ssl=config.verify_ssl) + record = {"domain_name": "nuget.acme.com", "domain_scope": "WORKSPACE"} + return [{**record, "domain_type": "NATIVE_NUGET"}] + + with ( + patch( + "cloudsmith_cli.credential_helpers.domains.get_cache_path", + return_value=tmp_path / "acme.json", + ), + patch("cloudsmith_cli.core.api.workspaces.list_domains", list_domains), + ): + args = ["-k", "k_abc", "-w", "acme", "--api-proxy", "http://proxy:3128", "-S"] + sent = _run_cli( + runner, [*args, "--", "-Plugin"], _credentials_request(CUSTOM_FEED) + ) + + assert seen == {"proxy": "http://proxy:3128", "verify_ssl": False} + assert _responses(sent)["c"]["Payload"]["ResponseCode"] == "Success" + + +@pytest.mark.parametrize( + "extra_args", [[], ["--api-headers", "Authorization=user:pass"]] +) +def test_cli_writes_only_protocol_messages_to_stdout_with_debug(runner, extra_args): + """Debug output goes to stderr, so it cannot break the protocol stream. + + With --debug, http.client prints each API request with print(). + """ + + def lookup_with_http_debug(*args, **kwargs): + print("send: b'GET /v2/workspaces/acme/domains/ HTTP/1.1'") + return True + + args = ["--debug", "credential-helper", "nuget", "-k", "k_abc", *extra_args] + with patch( + "cloudsmith_cli.credential_helpers.nuget.runtime.is_cloudsmith_domain", + lookup_with_http_debug, + ): + sent = _run_cli( + runner, [*args, "--", "-Plugin"], _credentials_request(), command=main + ) + assert _responses(sent)["c"]["Payload"]["Password"] == "k_abc" + + +def test_cli_retries_with_a_freshly_resolved_credential(runner, credential): + retry = _credentials_request() + retry["RequestId"] = "r" + retry["Payload"]["IsRetry"] = True + renewed = CredentialResult(api_key="k_renewed", source_name="test") + with patch( + "cloudsmith_cli.cli.commands.credential_helper.nuget.CredentialProviderChain" + ) as chain: + chain.return_value.resolve.side_effect = [credential, renewed] + sent = _run_cli( + runner, ["-k", "k_abc", "--", "-Plugin"], _credentials_request(), retry + ) + + assert _responses(sent)["c"]["Payload"]["Password"] == "k_abc" + assert _responses(sent)["r"]["Payload"]["Password"] == "k_renewed" + + +def test_cli_resolves_the_credential_at_the_first_credentials_request(runner): + """The handshake does not wait for a slow credential, such as an OIDC exchange.""" + oidc = CredentialResult(api_key="k_oidc", source_name="test") + with ( + patch("cloudsmith_cli.cli.decorators.CredentialProviderChain") as chain, + patch( + "cloudsmith_cli.cli.commands.credential_helper.nuget.CredentialProviderChain", + chain, + ), + ): + chain.return_value.resolve.return_value = oidc + _run_cli(runner, ["--", "-Plugin"], HANDSHAKE) + assert not chain.return_value.resolve.called + + requests = (_credentials_request(), _credentials_request()) + sent = _run_cli(runner, ["--", "-Plugin"], *requests) + + assert chain.return_value.resolve.call_count == 1 + assert _responses(sent)["c"]["Payload"]["Password"] == "k_oidc" + + +@pytest.mark.parametrize("non_interactive", [True, False]) +def test_cli_passes_the_interaction_policy_to_the_keyring(runner, non_interactive): + """When NuGet forbids prompts, the keyring does not ask for its password.""" + request = _credentials_request() + request["Payload"]["IsNonInteractive"] = non_interactive + no_prompt = [] + with patch( + "cloudsmith_cli.cli.commands.credential_helper.nuget.CredentialProviderChain" + ) as chain: + chain.return_value.resolve.side_effect = lambda context: no_prompt.append( + core_keyring.current_settings().no_prompt + ) + _run_cli(runner, ["--", "-Plugin"], request) + + assert no_prompt == [non_interactive] + + +def test_cli_keeps_the_resolved_credential_on_the_options(runner): + """Telemetry redacts ``opts.credential``, so the late credential goes there.""" + _run_cli(runner, ["-k", "k_abc", "--", "-Plugin"], _credentials_request()) + assert OPTIONS.value.credential.api_key == "k_abc" + + +def test_cli_matches_explicit_domains(runner): + args = ["-k", "k_abc", "-w", "acme", "--domain", "NuGet.Acme.com", "--no-discover"] + with patch(FORMAT_DOMAINS, side_effect=AssertionError("looked up")): + sent = _run_cli( + runner, [*args, "--", "-Plugin"], _credentials_request(CUSTOM_FEED) + ) + + assert _responses(sent)["c"]["Payload"]["Password"] == "k_abc" + + +def test_cli_without_nuget_args_prints_usage(runner): + result = runner.invoke(nuget, args=["-k", "k_abc"], catch_exceptions=False) + assert result.exit_code == 1 + assert result.stdout == "" + assert "credential-helper install nuget" in result.stderr + + +@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)) + for name in [name for name in os.environ if name.startswith("CLOUDSMITH_")]: + monkeypatch.delenv(name) + monkeypatch.delattr(sys, "frozen", raising=False) + return path + + +def test_installer_writes_the_launcher(bin_dir): + actions = NuGetInstaller().install(bin_dir=str(bin_dir)) + + assert (bin_dir / LAUNCHER).read_text() == ( + '#!/usr/bin/env sh\nexec cloudsmith credential-helper nuget -- "$@"\n' + ) + assert actions == [f"wrote launcher {bin_dir / LAUNCHER}"] + + +def test_installer_dry_run_and_uninstall_of_a_missing_launcher(bin_dir): + installer = NuGetInstaller() + + assert installer.install(bin_dir=str(bin_dir), dry_run=True) == [ + f"would write launcher {bin_dir / LAUNCHER}" + ] + assert installer.uninstall(bin_dir=str(bin_dir)) == [ + f"launcher not found at {bin_dir / LAUNCHER} (nothing to remove)" + ] + + +@pytest.fixture() +def on_windows(): + with ( + patch( + "cloudsmith_cli.credential_helpers.launchers.is_windows", return_value=True + ), + patch( + "cloudsmith_cli.credential_helpers.nuget.installer.is_windows", + return_value=True, + ), + ): + yield + + +@pytest.mark.usefixtures("on_windows") +def test_installer_uses_a_bat_launcher_on_windows(bin_dir): + """NuGet only discovers .exe and .bat plugins on Windows, not .cmd.""" + installer = NuGetInstaller() + installer.install( + bin_dir=str(bin_dir), + domains=("nuget.acme.com",), + launcher_env={"CLOUDSMITH_WORKSPACE": "acme"}, + ) + assert (bin_dir / f"{LAUNCHER}.bat").read_bytes() == ( + b"@echo off\r\n" + b"setlocal DisableDelayedExpansion\r\n" + b'set "CLOUDSMITH_WORKSPACE=acme"\r\n' + b'cloudsmith credential-helper nuget --domain "nuget.acme.com" -- %*\r\n' + ) + installer.uninstall(bin_dir=str(bin_dir)) + assert not (bin_dir / f"{LAUNCHER}.bat").exists() + + +def test_installer_passes_explicit_domains_to_the_plugin(bin_dir): + """NuGet sends every feed to the plugin, so the launcher names the hosts.""" + NuGetInstaller().install( + bin_dir=str(bin_dir), domains=("nuget.acme.com", "a b"), discover=False + ) + assert ( + "exec cloudsmith credential-helper nuget --domain nuget.acme.com " + "--domain 'a b' --no-discover -- \"$@\"" + ) in (bin_dir / LAUNCHER).read_text() + + +def test_list_command_reports_the_launcher(runner, bin_dir, monkeypatch): + """`list` finds the launcher where `install` puts it by default.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: bin_dir.parent)) + before = runner.invoke(list_cmd, args=["-F", "json"], catch_exceptions=False) + runner.invoke(install_cmd, args=["nuget"], catch_exceptions=False) + after = runner.invoke(list_cmd, args=["-F", "json"], catch_exceptions=False) + + def nuget_entry(result): + data = json.loads(result.stdout)["data"] + return next(entry for entry in data if entry["helper"] == "nuget") + + default_dir = bin_dir.parent / ".cloudsmith" / "bin" + assert nuget_entry(before)["launcher"] is None + assert nuget_entry(after)["launcher"] == str(default_dir / LAUNCHER) + assert nuget_entry(after)["hosts"] == ["all Cloudsmith NuGet feeds (nuget-plugin)"] + + +def test_install_command_keeps_install_time_settings(runner, bin_dir): + result = runner.invoke( + install_cmd, + args=["nuget", "--bin-dir", str(bin_dir), "-w", "acme"], + env={"CLOUDSMITH_API_KEY": "k_secret"}, + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert (bin_dir / LAUNCHER).read_text() == ( + "#!/usr/bin/env sh\nexport CLOUDSMITH_WORKSPACE=acme\n" + 'exec cloudsmith credential-helper nuget -- "$@"\n' + ) diff --git a/cloudsmith_cli/core/api/init.py b/cloudsmith_cli/core/api/init.py index b74b364f..50b91129 100644 --- a/cloudsmith_cli/core/api/init.py +++ b/cloudsmith_cli/core/api/init.py @@ -63,11 +63,11 @@ def initialise_api( config.api_key.pop("X-Api-Key", None) config.headers["Authorization"] = f"Bearer {credential.api_key}" if config.debug: - click.echo("SAML access token config value set") + click.echo("SAML access token config value set", err=True) else: config.api_key["X-Api-Key"] = credential.api_key if config.debug: - click.echo("User API key config value set") + click.echo("User API key config value set", err=True) auth_header = headers and config.headers.get("Authorization") if auth_header and " " in auth_header: @@ -76,7 +76,7 @@ def initialise_api( config.username, config.password = _basic_auth_credentials(encoded) if config.debug: - click.echo("Username and password config values set") + click.echo("Username and password config values set", err=True) # Important! Some of the attributes set above (e.g. error_retry_max) are not # present in the cloudsmith_api.Configuration class declaration. diff --git a/cloudsmith_cli/core/keyring.py b/cloudsmith_cli/core/keyring.py index d2c109e2..5564bab8 100644 --- a/cloudsmith_cli/core/keyring.py +++ b/cloudsmith_cli/core/keyring.py @@ -12,13 +12,14 @@ @dataclasses.dataclass(frozen=True) class KeyringSettings: - """The CLOUDSMITH_*KEYRING* settings that the CLI parsed.""" + """The CLOUDSMITH_*KEYRING* settings that the CLI parsed, and the prompt policy.""" no_keyring: bool = False backend: str | None = None key: str | None = None file_path: str | None = None directory: str | None = None + no_prompt: bool = False class _Current: @@ -38,6 +39,12 @@ def current_settings(): return _Current.settings +def forbid_password_prompts(): + """Make the keyring backend refuse to prompt for a password in this process.""" + if not _Current.settings.no_prompt: + configure(dataclasses.replace(_Current.settings, no_prompt=True)) + + def should_use_keyring(): """Check if keyring should be used based on CLOUDSMITH_NO_KEYRING.""" return not _Current.settings.no_keyring @@ -139,7 +146,9 @@ def _disable_password_prompts(backend): key. A chain calls a member only when the earlier members did not handle the call, so the error occurs only for a member that is in use. """ - if is_interactive(os.environ, controlling_terminal_mode()): + if not _Current.settings.no_prompt and is_interactive( + os.environ, controlling_terminal_mode() + ): return for member in getattr(backend, "backends", None) or [backend]: if hasattr(member, "_unlock"): diff --git a/cloudsmith_cli/core/tests/test_keyring.py b/cloudsmith_cli/core/tests/test_keyring.py index d2d031d6..ffee41e4 100644 --- a/cloudsmith_cli/core/tests/test_keyring.py +++ b/cloudsmith_cli/core/tests/test_keyring.py @@ -951,6 +951,15 @@ def test_supplied_key_uses_keyring_without_terminal( assert store_oidc_token(*OIDC_ARGS, "token-data") is True assert get_oidc_token(*OIDC_ARGS) == "token-data" + def test_forbidden_prompts_skip_keyring_at_a_terminal( + self, tmp_path, mock_get_user + ): + """A caller that must not prompt, such as NuGet, stops a terminal prompt too.""" + core_keyring.forbid_password_prompts() + backend = CryptFileKeyring() + with _keyring_session(backend, _keyring_env(tmp_path), has_terminal=True): + assert store_oidc_token(*OIDC_ARGS, "token-data") is False + def test_wrong_key_reports_not_found(self, tmp_path, mock_get_user): _write_oidc_entry(CryptFileKeyring(), tmp_path) core_keyring.configure(KeyringSettings(key="wrong-password")) diff --git a/cloudsmith_cli/credential_helpers/common.py b/cloudsmith_cli/credential_helpers/common.py index d6a0e98b..b29dbcc8 100644 --- a/cloudsmith_cli/credential_helpers/common.py +++ b/cloudsmith_cli/credential_helpers/common.py @@ -92,7 +92,7 @@ def is_standard_cloudsmith_domain(url): def is_cloudsmith_domain( - url, credential=None, api_host=None, backend_kind=None, org=None + url, credential=None, api_host=None, backend_kind=None, org=None, configure_api=True ): """ Check if a URL points to a Cloudsmith service. @@ -110,6 +110,8 @@ def is_cloudsmith_domain( org: Workspace slug whose custom domains to match against, as the CLI resolved it from --workspace, CLOUDSMITH_WORKSPACE, CLOUDSMITH_ORG, workspace in config.ini, or a legacy config alias + configure_api: Passed to the domain lookup. Pass ``False`` when the + caller has already configured the SDK. Returns: bool: True if this is a Cloudsmith domain @@ -137,13 +139,17 @@ def is_cloudsmith_domain( backend_kind, credential=credential, api_host=api_host, + configure_api=configure_api, ) } else: hosts = { d.host.lower() for d in get_workspace_domains( - org, credential=credential, api_host=api_host + org, + credential=credential, + api_host=api_host, + configure_api=configure_api, ) } return hostname in hosts diff --git a/cloudsmith_cli/credential_helpers/launchers.py b/cloudsmith_cli/credential_helpers/launchers.py index a3fc42da..28e03cdf 100644 --- a/cloudsmith_cli/credential_helpers/launchers.py +++ b/cloudsmith_cli/credential_helpers/launchers.py @@ -24,9 +24,9 @@ def is_windows() -> bool: return os.name == "nt" -def _launcher_filename(name: str, *, windows: bool) -> str: +def launcher_filename(name: str, *, windows: bool, windows_suffix: str = ".cmd") -> str: """Return the launcher file name for the platform (``.cmd`` on Windows).""" - return f"{name}.cmd" if windows else name + return f"{name}{windows_suffix}" if windows else name # A batch file expands % even inside quotes and cannot hold " or a line break. @@ -119,6 +119,7 @@ def write_launcher( name: str, target_cmd: str, dry_run=False, + windows_suffix=".cmd", env: dict[str, str] | None = None, ) -> Path: """Write a launcher script for *name* in *bin_dir* that execs *target_cmd*. @@ -132,6 +133,8 @@ def write_launcher( target_cmd: The command the launcher forwards to (e.g. ``cloudsmith credential-helper docker``). + windows_suffix: + The launcher file extension on Windows. NuGet discovers only ``.bat``. env: Environment variables that the launcher sets. @@ -141,16 +144,14 @@ def write_launcher( The path of the written file. """ windows = is_windows() + filename = launcher_filename(name, windows=windows, windows_suffix=windows_suffix) content = _launcher_content(target_cmd, windows=windows, env=env) if dry_run: - if windows: - return bin_dir / f"{name}.cmd" - else: - return bin_dir / name + return bin_dir / filename bin_dir = Path(bin_dir) bin_dir.mkdir(parents=True, exist_ok=True) - dest = bin_dir / _launcher_filename(name, windows=windows) + dest = bin_dir / filename dest.write_text(content, encoding="utf-8", newline="") if not windows: dest.chmod(0o755) @@ -158,7 +159,9 @@ def write_launcher( return dest -def remove_launcher(bin_dir: Path, name: str, dry_run=False) -> bool: +def remove_launcher( + bin_dir: Path, name: str, dry_run=False, windows_suffix=".cmd" +) -> bool: """Remove a launcher previously created by :func:`write_launcher`. Parameters @@ -167,13 +170,17 @@ def remove_launcher(bin_dir: Path, name: str, dry_run=False) -> bool: Directory that contains (or contained) the launcher. name: Base name of the helper binary (without extension). + windows_suffix: + The launcher file extension on Windows. Returns ------- bool ``True`` if a file was removed, ``False`` if no file was found. """ - target = Path(bin_dir) / _launcher_filename(name, windows=is_windows()) + target = Path(bin_dir) / launcher_filename( + name, windows=is_windows(), windows_suffix=windows_suffix + ) if target.exists(): if not dry_run: diff --git a/cloudsmith_cli/credential_helpers/nuget/__init__.py b/cloudsmith_cli/credential_helpers/nuget/__init__.py new file mode 100644 index 00000000..ac0d1d6a --- /dev/null +++ b/cloudsmith_cli/credential_helpers/nuget/__init__.py @@ -0,0 +1,4 @@ +# Copyright 2026 Cloudsmith Ltd +from .runtime import execute + +__all__ = ["execute"] diff --git a/cloudsmith_cli/credential_helpers/nuget/installer.py b/cloudsmith_cli/credential_helpers/nuget/installer.py new file mode 100644 index 00000000..2a75bda7 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/nuget/installer.py @@ -0,0 +1,147 @@ +# Copyright 2026 Cloudsmith Ltd +"""Installer for the NuGet credential provider. + +Writes a ``nuget-plugin-cloudsmith`` launcher into a directory on ``PATH``. +NuGet 6.13+ (.NET SDK 9.0.200+, Visual Studio 17.13+) discovers any +executable named ``nuget-plugin-*`` on ``PATH`` as an authentication plugin, +so no NuGet configuration file changes. On Windows NuGet only discovers +``.exe`` and ``.bat`` files, so the launcher is a ``.bat``. +""" + +from __future__ import annotations + +from ..common import path_warning +from ..launchers import ( + helper_command, + is_on_path, + is_windows, + launcher_arg, + launcher_filename, + remove_launcher, + resolve_bin_dir, + write_launcher, +) + + +class NuGetInstaller: + """Manages installation of the NuGet credential provider for Cloudsmith.""" + + LAUNCHER_NAME = "nuget-plugin-cloudsmith" + WINDOWS_SUFFIX = ".bat" + # NuGet can run the plugin from an IDE or a build agent without the + # install's environment, so the launcher keeps the install-time settings. + KEEPS_INSTALL_TIME_SETTINGS = True + + name = "nuget" + summary = "NuGet credential provider for Cloudsmith feeds" + + @classmethod + def _resolve_target_cmd(cls, domains: tuple[str, ...], discover: bool) -> str: + """Return the command the launcher forwards to. + + ``--`` stops the CLI from parsing NuGet's ``-Plugin`` argument as its + own option. + """ + args = [f"--domain {launcher_arg('--domain', domain)}" for domain in domains] + if not discover: + args.append("--no-discover") + return " ".join([helper_command(cls.name), *args, "--"]) + + def _launcher_path(self, target_dir): + return target_dir / launcher_filename( + self.LAUNCHER_NAME, windows=is_windows(), windows_suffix=self.WINDOWS_SUFFIX + ) + + def install( + self, + *, + bin_dir: str | None = None, + domains: tuple[str, ...] = (), + discover: bool = True, + launcher_env: dict[str, str] | None = None, + dry_run: bool = False, + # Accepted for a uniform installer interface. + **kwargs, + ) -> list[str]: + """Install the NuGet credential provider. + + Parameters + ---------- + bin_dir: + Override for the directory to install the launcher. It must be on + ``PATH`` for NuGet to discover the plugin. + domains: + Feed hosts that the plugin always authenticates. NuGet sends + every feed to the plugin, so the launcher passes them to it. + discover: + When ``False``, the plugin does not look up the Workspace's NuGet + domains through the Cloudsmith API. + launcher_env: + ``CLOUDSMITH_*`` settings that the launcher sets. + 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) + launcher_path = write_launcher( + target_dir, + self.LAUNCHER_NAME, + self._resolve_target_cmd(domains, discover), + dry_run=dry_run, + windows_suffix=self.WINDOWS_SUFFIX, + env=launcher_env, + ) + actions = [ + ( + f"would write launcher {launcher_path}" + if dry_run + else f"wrote launcher {launcher_path}" + ) + ] + if launcher_env: + actions.append(f"launcher sets {', '.join(launcher_env)}") + + if not is_on_path(target_dir): + actions.append( + path_warning(target_dir, "NuGet", self.LAUNCHER_NAME, is_windows()) + ) + + return actions + + def uninstall( + self, *, bin_dir: str | None = None, dry_run: bool = False + ) -> list[str]: + """Remove the NuGet credential provider launcher.""" + target_dir = resolve_bin_dir(bin_dir) + launcher_path = self._launcher_path(target_dir) + + removed = remove_launcher( + target_dir, + self.LAUNCHER_NAME, + dry_run=dry_run, + windows_suffix=self.WINDOWS_SUFFIX, + ) + if not removed: + return [f"launcher not found at {launcher_path} (nothing to remove)"] + if dry_run: + return [f"would remove launcher {launcher_path}"] + return [f"removed launcher {launcher_path}"] + + def status(self) -> dict: + """Return current installation status. + + The plugin is host-agnostic: NuGet sends it every feed URL, so + ``hosts`` is a single marker when the launcher exists. + """ + # resolve_bin_dir() gives ~/.cloudsmith/bin, where `install` writes by default. + launcher_path = self._launcher_path(resolve_bin_dir()) + if not launcher_path.exists(): + return {"launcher": None, "hosts": []} + return { + "launcher": str(launcher_path), + "hosts": ["all Cloudsmith NuGet feeds (nuget-plugin)"], + } diff --git a/cloudsmith_cli/credential_helpers/nuget/runtime.py b/cloudsmith_cli/credential_helpers/nuget/runtime.py new file mode 100644 index 00000000..ccc18b58 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/nuget/runtime.py @@ -0,0 +1,486 @@ +# Copyright 2026 Cloudsmith Ltd +""" +NuGet credential provider runtime. + +Transport-light protocol logic for NuGet credential providers. This module is +intentionally free of Click/sys imports so it can be unit-tested without +invoking the CLI machinery. + +NuGet (``dotnet``, MSBuild and Visual Studio) runs the plugin with +``-Plugin`` and speaks the cross-platform authentication plugin protocol: a +bidirectional conversation of newline-delimited JSON messages on stdin/stdout +that opens with a symmetric handshake, where each side sends a ``Handshake`` +request and answers the other's. + +See: + https://learn.microsoft.com/en-us/nuget/reference/extensibility/nuget-cross-platform-plugins +""" + +import collections +import json +import logging +import queue +import re +import threading +import time +import uuid +from urllib.parse import urlsplit + +from ..backends import BackendKind +from ..common import ( + REFUSAL_MESSAGE, + extract_hostname, + is_cloudsmith_domain, + is_standard_cloudsmith_domain, +) + +logger = logging.getLogger(__name__) + +#: Plugin protocol version spoken; authentication needs at least 2.0.0. +PROTOCOL_VERSION = "2.0.0" +MINIMUM_PROTOCOL_VERSION = "2.0.0" + +#: The request timeout in seconds until NuGet sends ``Initialize``. +DEFAULT_REQUEST_TIMEOUT = 5.0 + +#: A .NET TimeSpan in the constant format, ``[d.]hh:mm:ss[.fffffff]``. +_TIMESPAN = re.compile(r"(?:(\d+)\.)?(\d+):(\d+):(\d+(?:\.\d+)?)") + +_END_OF_INPUT = object() + +_NOT_APPLICABLE_MESSAGE = "Not a Cloudsmith NuGet feed." + +USAGE_MESSAGE = ( + "This is the Cloudsmith NuGet credential provider. NuGet runs it with " + "-Plugin once it is installed with `cloudsmith credential-helper install nuget`." +) + + +def is_plugin_mode(args) -> bool: + """Return True when NuGet asked for the plugin protocol (``-Plugin``).""" + return any(arg.lstrip("-/").lower() == "plugin" for arg in args) + + +def is_supported_source( + uri, + credential=None, + api_host=None, + workspace=None, + domains=frozenset(), + discover=True, +) -> bool: + """Return True when *uri* is a Cloudsmith NuGet feed. + + Standard ``*.cloudsmith.io``/``*.cloudsmith.com`` hosts always match. + Other hosts match when the Workspace lists them as NuGet domains (looked + up via the Cloudsmith API and cached, with the SDK as the caller configured + it). Only ``https`` feeds match, so the API key is never sent in clear + text. The host comes from the URI's authority, so a query or fragment + cannot name a Cloudsmith host. A backslash is refused because .NET reads + it as a path separator. The hosts in *domains* always match. When + *discover* is False, no other custom domain matches and the API is not + called. + """ + if not isinstance(uri, str) or "\\" in uri: + return False + try: + parts = urlsplit(uri) + except ValueError: + return False + if parts.scheme.lower() != "https" or not parts.hostname: + return False + if parts.hostname in domains: + return True + if not discover: + return is_standard_cloudsmith_domain(parts.hostname) + + return is_cloudsmith_domain( + parts.hostname, + credential=credential, + api_host=api_host, + backend_kind=BackendKind.NUGET, + org=workspace, + configure_api=False, + ) + + +def _parse_version(value) -> tuple[int, ...] | None: + """Parse a ``major.minor.patch`` SemVer string, ignoring any suffix.""" + if not isinstance(value, str): + return None + core = value.split("-", 1)[0].split("+", 1)[0] + try: + return tuple(int(part) for part in core.split(".")) + except ValueError: + return None + + +def parse_timespan(value) -> float | None: + """Return the seconds in a .NET TimeSpan string such as ``00:00:05``, or None.""" + match = _TIMESPAN.fullmatch(value) if isinstance(value, str) else None + if match is None: + return None + days, hours, minutes, seconds = match.groups() + return ((int(days or 0) * 24 + int(hours)) * 60 + int(minutes)) * 60 + float( + seconds + ) + + +def negotiate_version(payload) -> str | None: + """Return the protocol version for a Handshake request, or None if incompatible.""" + if not isinstance(payload, dict): + return None + theirs = _parse_version(payload.get("ProtocolVersion")) + their_minimum = _parse_version(payload.get("MinimumProtocolVersion")) + if theirs is None or their_minimum is None: + return None + ours = tuple(int(part) for part in PROTOCOL_VERSION.split(".")) + if not their_minimum <= ours <= theirs: + return None + return PROTOCOL_VERSION + + +def _is_cancel(message, request_id) -> bool: + """Return True when *message* cancels the request *request_id*.""" + return ( + isinstance(message, dict) + and message.get("Type") == "Cancel" + and message.get("RequestId") == request_id + ) + + +def _ends_session(message) -> bool: + """Return True for input that ends the session: Close, end of input or an error.""" + if message is _END_OF_INPUT or isinstance(message, Exception): + return True + return ( + isinstance(message, dict) + and message.get("Type") == "Request" + and message.get("Method") == "Close" + and bool(message.get("RequestId")) + ) + + +class _Answer: + """A worker's answer to one request. It stays a Fault if the request fails.""" + + def __init__(self): + self.message_type = "Fault" + self.payload = {"Message": "The plugin could not answer the request."} + + +class PluginSession: + """One NuGet plugin conversation over newline-delimited JSON streams.""" + + def __init__( + self, + stdin, + stdout, + credential=None, + api_host=None, + workspace=None, + refresh_credential=None, + domains=(), + discover=True, + ): + self.stdin = stdin + self.stdout = stdout + self.credential = credential + self.api_host = api_host + self.workspace = workspace + self.refresh_credential = refresh_credential + self._credential_resolved = credential is not None + self.domains = frozenset(extract_hostname(domain) for domain in domains) + self.discover = discover + self.refused = False + self.request_timeout = DEFAULT_REQUEST_TIMEOUT + self._handshake_id = str(uuid.uuid4()) + self._inbox = queue.Queue() + self._backlog = collections.deque() + # One request runs at a time, also after a Cancel. + self._request_lock = threading.Lock() + + def _send(self, request_id, message_type, method, payload=None) -> None: + """Write one message and flush it; NuGet reads line by line.""" + message = {"RequestId": request_id, "Type": message_type, "Method": method} + if payload is not None: + message["Payload"] = payload + self.stdout.write(json.dumps(message) + "\n") + self.stdout.flush() + + @staticmethod + def _handle_handshake(payload) -> dict: + version = negotiate_version(payload) + if version is None: + return {"ResponseCode": "Error"} + return {"ResponseCode": "Success", "ProtocolVersion": version} + + @staticmethod + def _handle_operation_claims(payload) -> dict: + # An authentication plugin is queried with no package source; a query + # for a specific source asks about package download, which this + # plugin does not offer. + payload = payload if isinstance(payload, dict) else {} + if payload.get("PackageSourceRepository") or payload.get("ServiceIndex"): + return {"Claims": []} + return {"Claims": ["Authentication"]} + + def _handle_initialize(self, payload) -> dict: + payload = payload if isinstance(payload, dict) else {} + request_timeout = parse_timespan(payload.get("RequestTimeout")) + if request_timeout: + self.request_timeout = request_timeout + return {"ResponseCode": "Success"} + + def _handle_get_credentials(self, payload) -> dict: + payload = payload if isinstance(payload, dict) else {} + uri = payload.get("Uri") + # Without discovery the host check needs no credential, so another + # provider's feed does not wait for one. + if not self.discover and not is_supported_source( + uri, domains=self.domains, discover=False + ): + return {"ResponseCode": "NotFound", "Message": _NOT_APPLICABLE_MESSAGE} + if self.refresh_credential and ( + payload.get("IsRetry") or not self._credential_resolved + ): + # The first request resolves a credential that the caller did not give. + # NuGet retries after the feed rejected the last credential, which + # can have expired since the plugin started. + self.credential = self.refresh_credential( + # A plugin must not prompt unless NuGet allows it. + non_interactive=payload.get("IsNonInteractive", True) + ) + # A failed resolution is not kept, so a later request tries again. + self._credential_resolved = self.credential is not None + # A failed custom-domain lookup matches no host, so NuGet gets + # NotFound and tries its other credential sources. + if not is_supported_source( + uri, + credential=self.credential, + api_host=self.api_host, + workspace=self.workspace, + domains=self.domains, + discover=self.discover, + ): + return {"ResponseCode": "NotFound", "Message": _NOT_APPLICABLE_MESSAGE} + + if not self.credential or not self.credential.api_key: + self.refused = True + return {"ResponseCode": "Error", "Message": REFUSAL_MESSAGE} + + return { + "ResponseCode": "Success", + "Username": "token", + "Password": self.credential.api_key, + "AuthenticationTypes": ["Basic"], + } + + def handle_request(self, method, payload) -> dict | None: + """Return the response payload for a request, or None for a fault.""" + if method == "Handshake": + return self._handle_handshake(payload) + if method == "GetOperationClaims": + return self._handle_operation_claims(payload) + if method == "GetAuthenticationCredentials": + return self._handle_get_credentials(payload) + if method == "Initialize": + return self._handle_initialize(payload) + if method in ( + "SetLogLevel", + "SetCredentials", + "MonitorNuGetProcessExit", + ): + # Nothing to configure: credentials come from the CLI's own + # provider chain. + return {"ResponseCode": "Success"} + return None + + def _work(self, answer, method, payload) -> None: + """Answer one request in a daemon thread, which cannot delay the exit.""" + try: + with self._request_lock: + answer.payload = self.handle_request(method, payload) + answer.message_type = "Response" + finally: + self._inbox.put(answer) + + def _answer_with_progress(self, request_id, method, payload) -> None: + """Answer a request from a worker thread, with Progress before each deadline. + + A Cancel for the request ends the wait. Other messages wait in the backlog. + Close, end of input or a stdin error ends the session at once. + """ + answer = _Answer() + threading.Thread( + target=self._work, args=(answer, method, payload), daemon=True + ).start() + interval = self.request_timeout / 2 + progress_due = time.monotonic() + interval + while True: + try: + item = self._inbox.get( + timeout=max(0.0, progress_due - time.monotonic()) + ) + except queue.Empty: + self._send(request_id, "Progress", method, {}) + progress_due = time.monotonic() + interval + continue + if item is answer: + self._send(request_id, answer.message_type, method, answer.payload) + return + if _is_cancel(item, request_id): + self._send(request_id, "Cancel", method) + return + if _ends_session(item): + # NuGet has gone, so the session ends before the backlog. + self._backlog.appendleft(item) + return + self._backlog.append(item) + + def _read_messages(self) -> None: + """Queue each stdin message, then the end of input, for the main thread.""" + try: + for line in self.stdin: + line = line.strip() + if not line: + continue + try: + self._inbox.put(json.loads(line)) + except ValueError as exc: + logger.debug("Ignoring malformed NuGet message: %s", exc) + except (OSError, ValueError) as exc: + self._inbox.put(exc) + self._inbox.put(_END_OF_INPUT) + + def _next_message(self): + """Return the next message, or raise the error that ended stdin.""" + message = self._backlog.popleft() if self._backlog else self._inbox.get() + if isinstance(message, Exception): + raise message + return message + + def _dispatch(self, message) -> bool: + """Handle one decoded message; return False when the session ends.""" + if not isinstance(message, dict): + return True + + request_id = message.get("RequestId") + message_type = message.get("Type") + method = message.get("Method") + + if message_type == "Response": + if request_id == self._handshake_id: + payload = message.get("Payload") + code = ( + payload.get("ResponseCode") if isinstance(payload, dict) else None + ) + if code != "Success": + logger.debug("NuGet rejected the plugin handshake: %r", code) + return False + return True + + if message_type != "Request" or not request_id: + # Progress, Cancel and Fault messages need no answer. + return True + + if method == "Close": + return False + + if method == "GetAuthenticationCredentials": + # A domain lookup or a credential renewal can take longer than + # NuGet's request timeout. + self._answer_with_progress(request_id, method, message.get("Payload")) + elif (response := self.handle_request(method, message.get("Payload"))) is None: + self._send( + request_id, + "Fault", + method, + {"Message": f"Unsupported request method {method!r}"}, + ) + else: + self._send(request_id, "Response", method, response) + return True + + def run(self) -> tuple[int, str | None]: + """ + Run the conversation until NuGet sends Close or closes stdin. + + Returns: + A (exit_code, stderr_text) tuple. Outcomes are reported in-band; + the exit code only flags a session that refused a Cloudsmith feed + for lack of credentials, or that broke at the transport level. + """ + try: + # The handshake is symmetric: NuGet waits for ours as well as for + # our answer to its own. + self._send( + self._handshake_id, + "Request", + "Handshake", + { + "ProtocolVersion": PROTOCOL_VERSION, + "MinimumProtocolVersion": MINIMUM_PROTOCOL_VERSION, + }, + ) + + threading.Thread(target=self._read_messages, daemon=True).start() + while (message := self._next_message()) is not _END_OF_INPUT: + if not self._dispatch(message): + break + except (OSError, ValueError) as exc: + # Protocol boundary: a broken pipe, a closed stream or undecodable + # input ends the session cleanly rather than with a traceback. + # NuGet has gone away, so a credential message would mislead. + logger.debug("nuget plugin session failed: %s", exc, exc_info=True) + return (1, None) + + if self.refused: + return (1, REFUSAL_MESSAGE) + return (0, None) + + +def execute( + args, + stdin, + stdout, + credential=None, + api_host=None, + workspace=None, + refresh_credential=None, + domains=(), + discover=True, +) -> tuple[int, str | None]: + """ + Run the NuGet credential provider for the given NuGet arguments. + + Args: + args: The arguments NuGet passed (``["-Plugin"]``) + stdin: A text stream to read plugin messages from + stdout: A text stream to write plugin messages to + credential: Pre-resolved CredentialResult from the provider chain + api_host: Cloudsmith API host URL + workspace: Workspace slug whose custom domains to match against + refresh_credential: Callable that resolves the credential at the + first request when *credential* is None, and again when NuGet + retries a request. It takes the request's ``IsNonInteractive`` + value as ``non_interactive``. + domains: Feed hosts that always match + discover: When False, do not look up the Workspace's NuGet domains + + Returns: + A (exit_code, stderr_text) tuple. + """ + if not is_plugin_mode(args): + return (1, USAGE_MESSAGE) + + return PluginSession( + stdin, + stdout, + credential=credential, + api_host=api_host, + workspace=workspace, + refresh_credential=refresh_credential, + domains=domains, + discover=discover, + ).run()