diff --git a/CHANGELOG.md b/CHANGELOG.md index 8b06944c..0c9e691f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ## [Unreleased] +### 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`. + ### Changed - `cloudsmith domains list` and the credential helpers now get the domains from `GET /v2/workspaces/{workspace}/domains/` on the configured API host. This list includes the Cloudsmith-owned hosts, so the CLI no longer contains a list of Cloudsmith hostnames. The CLI keeps a separate domain cache for each API host. 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 ad980d0e..7c87edc8 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/manage.py +++ b/cloudsmith_cli/cli/commands/credential_helper/manage.py @@ -12,9 +12,16 @@ from pathlib import Path import click +from click.core import ParameterSource from cloudsmith_cli.credential_helpers.cargo.installer import CargoInstaller from cloudsmith_cli.credential_helpers.generic import PartialInstallError +from cloudsmith_cli.credential_helpers.launchers import ( + ENV_ONLY_SETTINGS, + UnsafeLauncherValue, + install_time_settings, +) +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, @@ -44,6 +51,7 @@ "pnpm": PNPMInstaller, "cargo": CargoInstaller, "terraform": TerraformInstaller, + "nuget": NuGetInstaller, } @@ -118,6 +126,61 @@ def _terraform_helper_args(ctx, opts, repo: str | None = None) -> tuple[str, ... return tuple(args) +def _env_only_setting_options(f): + """Add hidden options that read the settings in ``ENV_ONLY_SETTINGS``. + + These settings have no flag, but ``install`` reads them through Click like + every other setting that a launcher keeps. + """ + for env_var in reversed(ENV_ONLY_SETTINGS): + flag = env_var.removeprefix("CLOUDSMITH_").lower().replace("_", "-") + option = click.option( + f"--{flag}-env", env_var.lower(), envvar=env_var, hidden=True + ) + f = option(f) + return f + + +def _launcher_settings_from_cli(ctx, opts, env_only_settings) -> dict[str, str]: + """Return the ``CLOUDSMITH_*`` settings that a launcher keeps. + + Only values from a flag or an environment variable are kept, because the + helper reads the config files itself. Secrets such as the API key are + never kept. The proxy comes only from config, because a proxy URL can + hold credentials and other users can read the launcher. + """ + supplied = (ParameterSource.COMMANDLINE, ParameterSource.ENVIRONMENT) + + def from_flag_or_env(param: str, value): + return value if ctx.get_parameter_source(param) in supplied else None + + return install_time_settings( + { + "CLOUDSMITH_WORKSPACE": from_flag_or_env("org", opts.org), + "CLOUDSMITH_PROFILE": ctx.meta.get("profile"), + "CLOUDSMITH_CONFIG_FILE": ctx.meta.get("config_file"), + "CLOUDSMITH_CREDENTIALS_FILE": ctx.meta.get("creds_file"), + "CLOUDSMITH_API_HOST": from_flag_or_env("api_host", opts.api_host), + "CLOUDSMITH_WITHOUT_API_SSL_VERIFY": from_flag_or_env( + "without_api_ssl_verify", + "false" if opts.api_ssl_verify else "true", + ), + "CLOUDSMITH_OIDC_AUDIENCE": from_flag_or_env( + "oidc_audience", opts.oidc_audience + ), + "CLOUDSMITH_SERVICE_SLUG": from_flag_or_env( + "oidc_service_slug", opts.oidc_service_slug + ), + **{ + env_var: from_flag_or_env( + env_var.lower(), env_only_settings[env_var.lower()] + ) + for env_var in ENV_ONLY_SETTINGS + }, + } + ) + + _REPO_FLAGS = ("-r", "--repo", "--repository") @@ -200,6 +263,7 @@ def _terraform_next_steps(helper_args: tuple[str, ...]) -> list[str]: "list (as `-r `) so `terraform init` needs no CLOUDSMITH_REPO env " "var. Terraform does not pass the repository to a credentials helper.", ) +@_env_only_setting_options @common_cli_config_options @common_cli_output_options @common_api_auth_options @@ -215,6 +279,7 @@ def install_cmd( no_discover: bool, refresh: bool, repo: str | None, + **env_only_settings: str | None, ) -> None: """Install a credential helper launcher and configure the package manager. @@ -269,12 +334,18 @@ def install_cmd( helper_args = _terraform_helper_args(ctx, opts, repo) install_kwargs["helper_args"] = helper_args + if getattr(installer, "KEEPS_INSTALL_TIME_SETTINGS", False): + install_kwargs["launcher_env"] = _launcher_settings_from_cli( + ctx, opts, env_only_settings + ) + try: actions = installer.install(**install_kwargs) except ( TerraformrcConflictError, TerraformHelperExeNotFound, TerraformPluginDirError, + UnsafeLauncherValue, ) as exc: raise click.ClickException(str(exc)) except OSError as exc: 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..98a22770 --- /dev/null +++ b/cloudsmith_cli/cli/commands/credential_helper/nuget.py @@ -0,0 +1,94 @@ +# 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.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 + + +@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 +@initialise_api +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, + credential=opts.credential, + api_host=opts.api_host, + workspace=opts.org, + refresh_credential=lambda: CredentialProviderChain().resolve( + opts.credential_context + ), + 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 98a2a66f..9a7a13e7 100644 --- a/cloudsmith_cli/cli/decorators.py +++ b/cloudsmith_cli/cli/decorators.py @@ -586,6 +586,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) 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 5e55b71d..45c1a137 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py @@ -20,10 +20,12 @@ from ....core.credentials.models import CredentialResult from ....credential_helpers.docker.installer import DockerInstaller from ....credential_helpers.launchers import ( + UnsafeLauncherValue, _launcher_content, - _launcher_filename, _user_bin_dir, + install_time_settings, is_on_path, + launcher_filename, remove_launcher, resolve_bin_dir, write_launcher, @@ -87,7 +89,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 ( @@ -96,6 +98,62 @@ def test_launcher_filename_and_content( ) +@pytest.mark.parametrize( + "profile,value,expected", + [ + ("C:\\Users\\jo", "C:\\Users\\jo\\a b&c!", "C:\\Users\\jo\\a b&c!"), + ( + "C:\\Users\\José", + "c:\\users\\josé\\cs\\config.ini", + "%USERPROFILE%\\cs\\config.ini", + ), + ], +) +def test_windows_launcher_settings_are_ascii(monkeypatch, profile, value, expected): + """cmd.exe reads a batch file in the console code page, not UTF-8. + + Only a non-ASCII path under the user profile is written as %USERPROFILE%, + because another account that runs NuGet has a different profile. + """ + monkeypatch.setenv("USERPROFILE", profile) + content = _launcher_content("cmd", windows=True, env={"CLOUDSMITH_X": value}) + assert content == ( + "@echo off\r\nsetlocal DisableDelayedExpansion\r\n" + f'set "CLOUDSMITH_X={expected}"\r\ncmd %*\r\n' + ) + + +@pytest.mark.parametrize("value", ["%PATH%", "D:\\Données"]) +def test_windows_launcher_refuses_a_value_a_batch_file_cannot_hold(monkeypatch, value): + monkeypatch.setenv("USERPROFILE", "C:\\Users\\José") + with pytest.raises(UnsafeLauncherValue, match="CLOUDSMITH_PROFILE"): + _launcher_content("cmd", windows=True, env={"CLOUDSMITH_PROFILE": value}) + + +def test_install_time_settings_resolve_paths_from_the_given_values( + monkeypatch, tmp_path +): + """Only the given values count; the process environment is not read.""" + monkeypatch.chdir(tmp_path) + monkeypatch.setenv("KEYRING_ROOT", "/var/keyring") + monkeypatch.setenv("CLOUDSMITH_KEYRING_BACKEND", "not.from.the.environment") + + settings = install_time_settings( + { + "CLOUDSMITH_PROFILE": None, + "CLOUDSMITH_NO_KEYRING": "1", + "CLOUDSMITH_KEYRING_DIR": "$KEYRING_ROOT/store", + "CLOUDSMITH_KEYRING_FILE_PATH": "keyring.cfg", + } + ) + + assert settings == { + "CLOUDSMITH_NO_KEYRING": "1", + "CLOUDSMITH_KEYRING_DIR": "/var/keyring/store", + "CLOUDSMITH_KEYRING_FILE_PATH": str(tmp_path / "keyring.cfg"), + } + + # --------------------------------------------------------------------------- # 2. write_launcher / remove_launcher — end-to-end on the host platform # --------------------------------------------------------------------------- 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..0b9ed86f --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_credential_helper_nuget.py @@ -0,0 +1,510 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the `cloudsmith credential-helper nuget` command and installer.""" + +from __future__ import annotations + +import io +import json +import sys +from unittest.mock import patch + +import click.testing +import cloudsmith_api +import pytest + +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, +) +from ..commands.credential_helper.manage import install_cmd, list_cmd +from ..commands.credential_helper.nuget import nuget +from ..commands.main import main + +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}, + ) + + +def _session(*messages, credential=None, workspace=None): + """Run a plugin session over *messages*; return (code, stderr, responses).""" + stdin = io.StringIO("".join(json.dumps(m) + "\n" for m in messages)) + stdout = io.StringIO() + code, stderr = PluginSession( + stdin, 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) -> list[dict]: + """Run the CLI over *messages* and return the messages on stdout.""" + stdin = "".join(json.dumps(m) + "\n" for m in messages) + 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): + stdin = io.StringIO(f"{first_line}\n{json.dumps(_credentials_request())}\n") + stdout = io.StringIO() + with patch( + "cloudsmith_cli.credential_helpers.nuget.runtime.uuid.uuid4", + return_value="ours", + ): + PluginSession(stdin, stdout, credential=credential).run() + + sent = [json.loads(line) for line in stdout.getvalue().splitlines()] + assert {m["RequestId"]: m["Type"] for m in sent[1:]} == answered + + +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 = json.dumps( + _message("p", "SetCredentials", {"ProxyPassword": "\u00e9t\u00e9"}), + ensure_ascii=False, + ) + stdin = f"{proxy_login}\n{json.dumps(_credentials_request())}\n".encode("cp1252") + result = runner.invoke( + nuget, args=["-k", "k_abc", "-Plugin"], input=stdin, catch_exceptions=False + ) + + sent = [json.loads(line) for line in result.stdout.splitlines()] + 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): + retry = _credentials_request() + 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.return_value = renewed + sent = _run_cli(runner, ["-k", "k_abc", "--", "-Plugin"], retry) + + assert _responses(sent)["c"]["Payload"]["Password"] == "k_renewed" + + +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)) + 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() == ( + '#!/bin/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): + with patch( + "cloudsmith_cli.credential_helpers.nuget.installer.resolve_bin_dir", + return_value=bin_dir, + ): + before = runner.invoke(list_cmd, args=["-F", "json"], catch_exceptions=False) + NuGetInstaller().install(bin_dir=str(bin_dir)) + 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") + + assert nuget_entry(before)["launcher"] is None + assert nuget_entry(after)["launcher"] == str(bin_dir / LAUNCHER) + assert nuget_entry(after)["hosts"] == ["all Cloudsmith NuGet feeds (nuget-plugin)"] + + +def test_install_command_bakes_flag_and_environment_settings(runner, bin_dir, tmp_path): + """Settings from flags and CLOUDSMITH_* variables go into the launcher. + + A value read from a config file does not, because the plugin reads that + file itself. Secrets and the proxy, which comes only from config, never + go into the launcher. + """ + config = tmp_path / "config.ini" + config.write_text("[default]\napi_ssl_verify = true\n") + result = runner.invoke( + install_cmd, + args=[ + "nuget", + "--bin-dir", + str(bin_dir), + "-w", + "acme", + "--config-file", + str(config), + "--api-host", + "https://api.example.com", + "--api-proxy", + "http://proxy.example.com:3128", + ], + env={ + "CLOUDSMITH_API_KEY": "k_secret", + "CLOUDSMITH_OIDC_TOKEN": "oidc_secret", + "CLOUDSMITH_KEYRING_KEY": "keyring_secret", + "CLOUDSMITH_KEYRING_DIR": "/var/keyring", + "CLOUDSMITH_PROFILE": "work", + "CLOUDSMITH_OIDC_AUDIENCE": "cloudsmith-test", + "CLOUDSMITH_SERVICE_SLUG": "ci-bot", + "CLOUDSMITH_WITHOUT_API_SSL_VERIFY": "false", + }, + catch_exceptions=False, + ) + assert result.exit_code == 0, result.output + + launcher = (bin_dir / LAUNCHER).read_text() + exported = dict( + line.split("export ", 1)[1].split("=", 1) + for line in launcher.splitlines() + if "export " in line + ) + assert exported == { + "CLOUDSMITH_WORKSPACE": "acme", + "CLOUDSMITH_PROFILE": "work", + "CLOUDSMITH_CONFIG_FILE": str(config.resolve()), + "CLOUDSMITH_API_HOST": "https://api.example.com", + "CLOUDSMITH_OIDC_AUDIENCE": "cloudsmith-test", + "CLOUDSMITH_SERVICE_SLUG": "ci-bot", + "CLOUDSMITH_WITHOUT_API_SSL_VERIFY": "false", + "CLOUDSMITH_KEYRING_DIR": "/var/keyring", + } + assert "secret" not in launcher + assert "proxy" not in launcher diff --git a/cloudsmith_cli/core/api/init.py b/cloudsmith_cli/core/api/init.py index 8a373635..c9dc3fe8 100644 --- a/cloudsmith_cli/core/api/init.py +++ b/cloudsmith_cli/core/api/init.py @@ -48,11 +48,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("SSO 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: @@ -63,7 +63,7 @@ def initialise_api( config.username, config.password = values.split(":") 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/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 69572b09..3dfb9dc4 100644 --- a/cloudsmith_cli/credential_helpers/launchers.py +++ b/cloudsmith_cli/credential_helpers/launchers.py @@ -25,15 +25,81 @@ 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 -def launcher_arg(value: str) -> str: +# Settings with no CLI flag that change where a helper finds credentials. +ENV_ONLY_SETTINGS = ( + "CLOUDSMITH_NO_KEYRING", + "CLOUDSMITH_KEYRING_BACKEND", + "CLOUDSMITH_KEYRING_FILE_PATH", + "CLOUDSMITH_KEYRING_DIR", +) +_PATH_SETTINGS = ("CLOUDSMITH_KEYRING_FILE_PATH", "CLOUDSMITH_KEYRING_DIR") + +# A batch file expands % even inside quotes and cannot hold " or a line break. +_UNSAFE_IN_BATCH = ('"', "%", "\r", "\n") + + +class UnsafeLauncherValue(ValueError): + """Raised when a value cannot go into a launcher script.""" + + +def install_time_settings(supplied: dict[str, str | None]) -> dict[str, str]: + """Return the ``CLOUDSMITH_*`` settings that a launcher keeps. + + A package manager can run its helper from an IDE or a build agent that + does not have the environment of the install. *supplied* holds the + settings that the install got from a flag or an environment variable. + Paths are made absolute, because the helper runs in another directory. + """ + settings = dict(supplied) + for name in _PATH_SETTINGS: + path = settings.get(name) + if path: + settings[name] = os.path.abspath( + os.path.expanduser(os.path.expandvars(path)) + ) + return {name: value for name, value in settings.items() if value} + + +def batch_value(name: str, value: str) -> str: + """Return *value* in a form that a batch file holds without change. + + ``cmd.exe`` reads a batch file in the console code page, so the file must + be ASCII. A path under the user profile is written as ``%USERPROFILE%``, + which ``cmd.exe`` expands to the correct Unicode path when it runs. An + ASCII path stays literal, because another account that runs the package + manager has a different profile. + """ + if any(char in value for char in _UNSAFE_IN_BATCH): + raise UnsafeLauncherValue( + f"{name} {value!r} cannot go into the launcher: a batch file cannot " + "hold a '\"', a '%' or a line break." + ) + profile = os.environ.get("USERPROFILE", "") + lowered = value.lower() + if ( + not value.isascii() + and profile + and (lowered == profile.lower() or lowered.startswith(profile.lower() + "\\")) + ): + value = "%USERPROFILE%" + value[len(profile) :] + if not value.isascii(): + raise UnsafeLauncherValue( + f"{name} {value!r} cannot go into the launcher: outside the user " + "profile, a batch file can only hold ASCII characters. Set it in " + "the environment that runs the package manager instead." + ) + return value + + +def launcher_arg(name: str, value: str) -> str: """Return *value* quoted as one argument of a launcher command.""" if is_windows(): - return f'"{value}"' + return f'"{batch_value(name, value)}"' return shlex.quote(value) @@ -44,20 +110,36 @@ def helper_command(helper: str) -> str: ``cloudsmith``, so the launcher runs the absolute executable. """ if getattr(sys, "frozen", False): - return f"{launcher_arg(sys.executable)} credential-helper {helper}" + executable = launcher_arg("cloudsmith", sys.executable) + return f"{executable} credential-helper {helper}" return f"cloudsmith credential-helper {helper}" -def _launcher_content(target_cmd: str, *, windows: bool) -> str: +def _launcher_content( + target_cmd: str, *, windows: bool, env: dict[str, str] | None = None +) -> str: """Return the launcher script body for the platform. Windows uses a ``.cmd`` batch file (``@echo off`` keeps stdout clean for Docker's credential JSON); Unix uses a ``#!/bin/sh`` script that ``exec``s - the target so signals and the exit code pass straight through. + the target so signals and the exit code pass straight through. The + ``env`` variables replace the caller's values, so the helper gives the + same result as the install. """ + env = env or {} if windows: - return f"@echo off\r\n{target_cmd} %*\r\n" - return f'#!/bin/sh\nexec {target_cmd} "$@"\n' + settings = "".join( + f'set "{name}={batch_value(name, value)}"\r\n' + for name, value in env.items() + ) + if settings: + # Keep "!" literal and the settings local to the launcher. + settings = f"setlocal DisableDelayedExpansion\r\n{settings}" + return f"@echo off\r\n{settings}{target_cmd} %*\r\n" + settings = "".join( + f"export {name}={shlex.quote(value)}\n" for name, value in env.items() + ) + return f'#!/bin/sh\n{settings}exec {target_cmd} "$@"\n' def _user_bin_dir(windows: bool) -> Path: @@ -73,7 +155,14 @@ def _user_bin_dir(windows: bool) -> Path: return Path.home() / ".local" / "bin" -def write_launcher(bin_dir: Path, name: str, target_cmd: str, dry_run=False) -> Path: +def write_launcher( + bin_dir: Path, + 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*. Parameters @@ -85,6 +174,10 @@ def write_launcher(bin_dir: Path, name: str, target_cmd: str, dry_run=False) -> 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. Returns ------- @@ -92,25 +185,24 @@ def write_launcher(bin_dir: Path, name: str, target_cmd: str, dry_run=False) -> 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.write_text( - _launcher_content(target_cmd, windows=windows), encoding="utf-8", newline="" - ) + dest = bin_dir / filename + dest.write_text(content, encoding="utf-8", newline="") if not windows: dest.chmod(0o755) 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 @@ -119,13 +211,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..7198d5ed --- /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, from + :func:`~cloudsmith_cli.credential_helpers.launchers.install_time_settings`. + 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. + """ + 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..591a2580 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/nuget/runtime.py @@ -0,0 +1,343 @@ +# 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 json +import logging +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" + +_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 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 + + +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.domains = frozenset(extract_hostname(domain) for domain in domains) + self.discover = discover + self.refused = False + self._handshake_id = str(uuid.uuid4()) + + 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_get_credentials(self, payload) -> dict: + payload = payload if isinstance(payload, dict) else {} + if payload.get("IsRetry") and self.refresh_credential: + # NuGet retries after the feed rejected the last credential, which + # can have expired since the plugin started. + self.credential = self.refresh_credential() + uri = payload.get("Uri") + # 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 in ( + "Initialize", + "SetLogLevel", + "SetCredentials", + "MonitorNuGetProcessExit", + ): + # Nothing to configure: credentials come from the CLI's own + # provider chain and every request is answered immediately. + return {"ResponseCode": "Success"} + return None + + 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 + + response = self.handle_request(method, message.get("Payload")) + if response 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, + }, + ) + + for line in self.stdin: + line = line.strip() + if not line: + continue + try: + message = json.loads(line) + except ValueError as exc: + logger.debug("Ignoring malformed NuGet message: %s", exc) + continue + 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 again when + NuGet retries a request + 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()