From 5bb95cee21020c6a28ec0cadbe9b49772b8dcaca Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Wed, 12 Aug 2026 23:45:46 +0100 Subject: [PATCH 1/8] feat(issue-83): add `cloudsmith repos gpg` command group Adds CLI support for managing a repository's GPG signing key, closing the gap noted in cloudsmith-io/cloudsmith-cli#83: the API has supported this since repos_gpg_list/create/regenerate were added to the SDK, but the CLI never exposed it. - core/api/repos.py: list_repo_gpg_key, create_repo_gpg_key and regenerate_repo_gpg_key wrappers around the SDK's repos_gpg_* endpoints, translating SDK exceptions to ApiException per existing convention. - cli/commands/repos.py: `cloudsmith repos gpg get|upload|regenerate OWNER/REPO`. Key/passphrase material is only ever read from a file (or stdin via '-') or an interactive hide_input prompt, never a bare command-line flag, so it can't leak into shell history or the process list. `regenerate` asks for confirmation first (like `repos delete`), since it invalidates the repository's current key. - Tests: httpretty-mocked API tests in core/tests/test_repos.py, and mock-patched CLI tests in cli/tests/commands/test_repos.py. There is no delete/rotate-off endpoint on the backend, so there's no `delete` subcommand - `get`/`upload`/`regenerate` is the full surface. --- cloudsmith_cli/cli/commands/repos.py | 236 ++++++++++++++++++ .../cli/tests/commands/test_repos.py | 182 +++++++++++++- cloudsmith_cli/core/api/repos.py | 39 +++ cloudsmith_cli/core/tests/test_repos.py | 209 ++++++++++++++++ 4 files changed, 665 insertions(+), 1 deletion(-) create mode 100644 cloudsmith_cli/core/tests/test_repos.py diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index 4308abda..a40e1cd7 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -272,6 +272,242 @@ def update(ctx, opts, owner_repo, repo_config_file): print_repositories(opts=opts, data=[repository], show_list_info=True) +def print_gpg_key(gpg_key): + """Print a repository's GPG key details as human-readable text.""" + click.echo() + click.echo( + f"Fingerprint: {click.style(gpg_key.get('fingerprint') or '(none)', fg='green')}" + ) + click.echo( + "Fingerprint (short): " + f"{click.style(gpg_key.get('fingerprint_short') or '(none)', fg='green')}" + ) + click.echo(f"Active: {click.style(str(bool(gpg_key.get('active'))), fg='blue')}") + click.echo(f"Default: {click.style(str(bool(gpg_key.get('default'))), fg='blue')}") + + comment = gpg_key.get("comment") + if comment: + click.echo(f"Comment: {comment}") + + public_key = gpg_key.get("public_key") + if public_key: + click.echo() + click.echo("Public Key:") + click.echo(public_key) + + click.echo() + + +@repositories.group(cls=command.AliasGroup, name="gpg", aliases=[]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.pass_context +def gpg(ctx, opts): # pylint: disable=unused-argument + """ + Manage a repository's GPG signing key. + + See the help for subcommands for more information on each. + + Note: The API doesn't currently support deleting a repository's GPG key, + so there's no 'delete' subcommand here. + """ + + +@gpg.command(name="get", aliases=["list", "ls"]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo", metavar="OWNER/REPO", callback=validators.validate_owner_repo +) +@click.pass_context +def gpg_get(ctx, opts, owner_repo): + """ + Get the active GPG signing key for a repository. + + - OWNER/REPO: Specify the OWNER namespace (i.e. user or org), and the + REPO name to get the GPG key for, separated by a slash. + + Example: 'your-org/your-repo' + + Full CLI example: + + $ cloudsmith repos gpg get your-org/your-repo + """ + owner, repo = owner_repo + use_stderr = utils.should_use_stderr(opts) + + click.echo("Getting GPG key ... ", nl=False, err=use_stderr) + + context_msg = "Failed to get the repository GPG key!" + with ( + handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + maybe_spinner(opts), + ): + gpg_key = api.list_repo_gpg_key(owner, repo) + + click.secho("OK", fg="green", err=use_stderr) + + if utils.maybe_print_as_json(opts, gpg_key): + return + + print_gpg_key(gpg_key) + + +@gpg.command(name="upload", aliases=["set"]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo", metavar="OWNER/REPO", callback=validators.validate_owner_repo +) +@click.option( + "--private-key-file", + "private_key_file", + type=click.File("r"), + required=True, + help="Path to a file containing the armored GPG private key to upload. " + "Use '-' to read from stdin.", +) +@click.option( + "--passphrase-file", + "passphrase_file", + type=click.File("r"), + required=False, + default=None, + help="Path to a file containing the GPG private key's passphrase. If " + "omitted, you'll be prompted for it interactively (leave blank if the " + "key has none). The passphrase is never accepted as a plain " + "command-line value, to avoid leaking it into shell history or the " + "process list.", +) +@click.pass_context +def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): + """ + Set (upload) the active GPG signing key for a repository. + + - OWNER/REPO: Specify the OWNER namespace (i.e. user or org), and the + REPO name to set the GPG key for, separated by a slash. + + Example: 'your-org/your-repo' + + The private key material is always read from a file (or stdin via '-'), + never accepted as a plain command-line argument. + + Full CLI example: + + $ cloudsmith repos gpg upload your-org/your-repo --private-key-file key.asc + """ + owner, repo = owner_repo + use_stderr = utils.should_use_stderr(opts) + + gpg_private_key = private_key_file.read() + if not gpg_private_key.strip(): + raise click.BadParameter( + "The private key file is empty.", param_hint="--private-key-file" + ) + + if passphrase_file is not None: + gpg_passphrase = passphrase_file.read().strip() or None + else: + gpg_passphrase = ( + click.prompt( + "GPG passphrase (leave blank if the key has none)", + hide_input=True, + default="", + show_default=False, + ).strip() + or None + ) + + click.echo( + f"Uploading GPG key for {click.style(repo, bold=True)} in the " + f"{click.style(owner, bold=True)} namespace ... ", + nl=False, + err=use_stderr, + ) + + context_msg = "Failed to set the repository GPG key!" + with ( + handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + maybe_spinner(opts), + ): + gpg_key = api.create_repo_gpg_key( + owner, repo, gpg_private_key=gpg_private_key, gpg_passphrase=gpg_passphrase + ) + + click.secho("OK", fg="green", err=use_stderr) + + if utils.maybe_print_as_json(opts, gpg_key): + return + + print_gpg_key(gpg_key) + + +@gpg.command(name="regenerate", aliases=["regen"]) +@decorators.common_cli_config_options +@decorators.common_cli_output_options +@decorators.common_api_auth_options +@decorators.initialise_api +@click.argument( + "owner_repo", metavar="OWNER/REPO", callback=validators.validate_owner_repo +) +@click.option( + "-y", + "--yes", + default=False, + is_flag=True, + help="Assume yes as default answer to questions (this is dangerous!)", +) +@click.pass_context +def gpg_regenerate(ctx, opts, owner_repo, yes): + """ + Regenerate the GPG signing key for a repository. + + - OWNER/REPO: Specify the OWNER namespace (i.e. user or org), and the + REPO name to regenerate the GPG key for, separated by a slash. + + Example: 'your-org/your-repo' + + This replaces the repository's current GPG key with a newly generated + one; consumers relying on the old key's fingerprint will need to pick up + the new one. There is no way to undo this from the CLI. + + Full CLI example: + + $ cloudsmith repos gpg regenerate your-org/your-repo + """ + owner, repo = owner_repo + use_stderr = utils.should_use_stderr(opts) + + prompt = ( + f"regenerate the GPG key for {click.style(repo, bold=True)} in the " + f"{click.style(owner, bold=True)} namespace" + ) + if not utils.confirm_operation(prompt, assume_yes=yes, err=use_stderr): + return + + click.echo("Regenerating GPG key ... ", nl=False, err=use_stderr) + + context_msg = "Failed to regenerate the repository GPG key!" + with ( + handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + maybe_spinner(opts), + ): + gpg_key = api.regenerate_repo_gpg_key(owner, repo) + + click.secho("OK", fg="green", err=use_stderr) + + if utils.maybe_print_as_json(opts, gpg_key): + return + + print_gpg_key(gpg_key) + + @repositories.command(aliases=["rm"]) @decorators.common_cli_config_options @decorators.common_cli_output_options diff --git a/cloudsmith_cli/cli/tests/commands/test_repos.py b/cloudsmith_cli/cli/tests/commands/test_repos.py index 4d159167..95658b2d 100644 --- a/cloudsmith_cli/cli/tests/commands/test_repos.py +++ b/cloudsmith_cli/cli/tests/commands/test_repos.py @@ -1,11 +1,37 @@ import json +from unittest.mock import patch import pytest from ...commands.list_ import repos as list_repos -from ...commands.repos import create, delete, get, update +from ...commands.repos import ( + create, + delete, + get, + gpg_get, + gpg_regenerate, + gpg_upload, + update, +) from ..utils import random_str +HERMETIC_ARGS = ["--api-key", "fake-api-key"] + +# Not a real GPG key - deliberately not using the literal +# "-----BEGIN ... PRIVATE KEY-----" marker so secret-scanning tooling (e.g. +# the detect-private-key pre-commit hook) doesn't flag this fixture. +_FAKE_GPG_KEY_MATERIAL = "fake-armored-gpg-private-key-material-for-tests-only" + +_GPG_KEY = { + "active": True, + "comment": "my-repo GPG key", + "created_at": "2026-01-01T00:00:00Z", + "default": True, + "fingerprint": "AAAA1111BBBB2222CCCC3333DDDD4444EEEE5555", + "fingerprint_short": "EEEE5555", + "public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...\n-----END PGP PUBLIC KEY BLOCK-----", +} + def create_repo_config_file(directory, name, description, repository_type_str, slug): """Create a REPO-CONFIG.json file in `directory` with the values provided.""" @@ -200,3 +226,157 @@ def test_repos_commands(runner, organization, tmp_path): + " namespace ... OK" in result.output ) + + +class TestReposGpgGet: + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + def test_success_prints_fingerprint(self, mock_list, runner): + mock_list.return_value = dict(_GPG_KEY) + + result = runner.invoke( + gpg_get, ["my-org/my-repo", *HERMETIC_ARGS], catch_exceptions=False + ) + + assert result.exit_code == 0, result.output + mock_list.assert_called_once_with("my-org", "my-repo") + assert _GPG_KEY["fingerprint"] in result.output + assert _GPG_KEY["fingerprint_short"] in result.output + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + def test_json_output(self, mock_list, runner): + mock_list.return_value = dict(_GPG_KEY) + + result = runner.invoke( + gpg_get, + ["my-org/my-repo", "-F", "json", *HERMETIC_ARGS], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + # "Getting GPG key ... " progress text goes to stderr, but the runner + # merges streams, so pick out the JSON line specifically. + json_line = next( + line for line in result.output.splitlines() if line.startswith("{") + ) + document = json.loads(json_line) + assert document["data"]["fingerprint"] == _GPG_KEY["fingerprint"] + + def test_invalid_owner_repo_rejected(self, runner): + result = runner.invoke( + gpg_get, ["not-a-valid-argument", *HERMETIC_ARGS], catch_exceptions=False + ) + + assert result.exit_code != 0 + assert "Must be in the form of OWNER/REPO" in result.output + + +class TestReposGpgUpload: + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_uploads_key_and_passphrase_from_files(self, mock_create, runner, tmp_path): + mock_create.return_value = dict(_GPG_KEY) + + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + passphrase_file = tmp_path / "passphrase.txt" + passphrase_file.write_text("s3cret\n") + + result = runner.invoke( + gpg_upload, + [ + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--passphrase-file", + str(passphrase_file), + *HERMETIC_ARGS, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + mock_create.assert_called_once_with( + "my-org", + "my-repo", + gpg_private_key=_FAKE_GPG_KEY_MATERIAL, + gpg_passphrase="s3cret", + ) + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_prompts_for_passphrase_when_no_file_given( + self, mock_create, runner, tmp_path + ): + mock_create.return_value = dict(_GPG_KEY) + + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + gpg_upload, + ["my-org/my-repo", "--private-key-file", str(key_file), *HERMETIC_ARGS], + input="\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + mock_create.assert_called_once_with( + "my-org", + "my-repo", + gpg_private_key=_FAKE_GPG_KEY_MATERIAL, + gpg_passphrase=None, + ) + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_empty_private_key_file_rejected(self, mock_create, runner, tmp_path): + key_file = tmp_path / "key.asc" + key_file.write_text(" \n") + + result = runner.invoke( + gpg_upload, + ["my-org/my-repo", "--private-key-file", str(key_file), *HERMETIC_ARGS], + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "private key file is empty" in result.output + mock_create.assert_not_called() + + def test_private_key_flag_not_accepted(self, runner): + """Key material must never be a plain CLI value (shell-history/process-list leak).""" + result = runner.invoke( + gpg_upload, + ["my-org/my-repo", "--private-key", "sekrit", *HERMETIC_ARGS], + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "no such option" in result.output.lower() + + +class TestReposGpgRegenerate: + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_prompts_for_confirmation_and_declines(self, mock_regenerate, runner): + result = runner.invoke( + gpg_regenerate, + ["my-org/my-repo", *HERMETIC_ARGS], + input="N", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "OK, phew! Close call. :-)" in result.output + mock_regenerate.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_yes_flag_skips_confirmation(self, mock_regenerate, runner): + new_key = dict(_GPG_KEY, fingerprint="1111AAAA2222BBBB3333CCCC4444DDDD5555EEEE") + mock_regenerate.return_value = new_key + + result = runner.invoke( + gpg_regenerate, + ["my-org/my-repo", "-y", *HERMETIC_ARGS], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + mock_regenerate.assert_called_once_with("my-org", "my-repo") + assert new_key["fingerprint"] in result.output diff --git a/cloudsmith_cli/core/api/repos.py b/cloudsmith_cli/core/api/repos.py index 8591155d..dd9b3053 100644 --- a/cloudsmith_cli/core/api/repos.py +++ b/cloudsmith_cli/core/api/repos.py @@ -79,3 +79,42 @@ def delete_repo(owner, repo): _, _, headers = client.repos_delete_with_http_info(owner, repo) ratelimits.maybe_rate_limit(client, headers) + + +def list_repo_gpg_key(owner, repo): + """Get the active GPG key for a repository.""" + client = get_repos_api() + + with catch_raise_api_exception(): + data, _, headers = client.repos_gpg_list_with_http_info(owner, repo) + + ratelimits.maybe_rate_limit(client, headers) + return data.to_dict() + + +def create_repo_gpg_key(owner, repo, gpg_private_key, gpg_passphrase=None): + """Set (upload) the active GPG key for a repository.""" + client = get_repos_api() + + gpg_key_create = cloudsmith_api.RepositoryGpgKeyCreate( + gpg_private_key=gpg_private_key, gpg_passphrase=gpg_passphrase + ) + + with catch_raise_api_exception(): + data, _, headers = client.repos_gpg_create_with_http_info( + owner, repo, data=gpg_key_create + ) + + ratelimits.maybe_rate_limit(client, headers) + return data.to_dict() + + +def regenerate_repo_gpg_key(owner, repo): + """Regenerate the GPG key for a repository.""" + client = get_repos_api() + + with catch_raise_api_exception(): + data, _, headers = client.repos_gpg_regenerate_with_http_info(owner, repo) + + ratelimits.maybe_rate_limit(client, headers) + return data.to_dict() diff --git a/cloudsmith_cli/core/tests/test_repos.py b/cloudsmith_cli/core/tests/test_repos.py new file mode 100644 index 00000000..d4ffae72 --- /dev/null +++ b/cloudsmith_cli/core/tests/test_repos.py @@ -0,0 +1,209 @@ +"""Tests for the repository GPG key API client.""" + +import json + +import httpretty +import httpretty.core +import pytest + +from .. import keyring +from ..api import repos +from ..api.exceptions import ApiException +from ..api.init import initialise_api +from ..credentials.models import CredentialResult + +API_HOST = "https://api.cloudsmith.io" +OWNER = "my-org" +REPO = "my-repo" +GPG_URL = f"{API_HOST}/repos/{OWNER}/{REPO}/gpg/" +GPG_REGENERATE_URL = f"{API_HOST}/repos/{OWNER}/{REPO}/gpg/regenerate/" + +# Not a real GPG key - armor-shaped fixture content, deliberately not using +# the literal "-----BEGIN ... PRIVATE KEY-----" marker so secret-scanning +# tooling (e.g. the detect-private-key pre-commit hook) doesn't flag it. +FAKE_GPG_KEY_MATERIAL = "fake-armored-gpg-private-key-material-for-tests-only" + + +@pytest.fixture(autouse=True) +def _setup_api(monkeypatch): + """Initialise the SDK Configuration and stub keyring lookups. + + Mirrors the metadata API test setup: initialise_api() registers custom + retry attributes on cloudsmith_api.Configuration that create_requests_session + expects, and keyring is stubbed so we never touch the user's real SSO tokens + during a test run. + """ + monkeypatch.setattr(keyring, "get_access_token", lambda host: None) + monkeypatch.setattr(keyring, "get_refresh_token", lambda host: None) + monkeypatch.setattr(keyring, "should_refresh_access_token", lambda host: False) + monkeypatch.setattr( + httpretty.core.fakesock.socket, + "shutdown", + lambda self, how: None, + raising=False, + ) + initialise_api( + host=API_HOST, + credential=CredentialResult( + api_key="test-api-key", + source_name="test", + auth_type="api_key", + ), + ) + + +def _last_request(): + return httpretty.last_request() + + +def _gpg_key_body(**overrides): + body = { + "active": True, + "comment": "my-repo GPG key", + "created_at": "2026-01-01T00:00:00Z", + "default": True, + "fingerprint": "AAAA1111BBBB2222CCCC3333DDDD4444EEEE5555", + "fingerprint_short": "EEEE5555", + "public_key": "-----BEGIN PGP PUBLIC KEY BLOCK-----\n...\n-----END PGP PUBLIC KEY BLOCK-----", + } + body.update(overrides) + return body + + +class TestListRepoGpgKey: + @httpretty.activate(allow_net_connect=False) + def test_success_returns_gpg_key_dict(self): + body = _gpg_key_body() + httpretty.register_uri( + httpretty.GET, + GPG_URL, + body=json.dumps(body), + status=200, + content_type="application/json", + ) + + result = repos.list_repo_gpg_key(OWNER, REPO) + + assert result["fingerprint"] == body["fingerprint"] + assert result["fingerprint_short"] == body["fingerprint_short"] + assert result["active"] is True + assert result["default"] is True + + sent = _last_request() + assert sent.headers.get("X-Api-Key") == "test-api-key" + + @httpretty.activate(allow_net_connect=False) + def test_404_raises_api_exception(self): + httpretty.register_uri( + httpretty.GET, + GPG_URL, + body=json.dumps({"detail": "Not found."}), + status=404, + content_type="application/json", + ) + + with pytest.raises(ApiException) as exc_info: + repos.list_repo_gpg_key(OWNER, REPO) + + assert exc_info.value.status == 404 + + +class TestCreateRepoGpgKey: + @httpretty.activate(allow_net_connect=False) + def test_success_sends_private_key_and_passphrase(self): + body = _gpg_key_body() + httpretty.register_uri( + httpretty.POST, + GPG_URL, + body=json.dumps(body), + status=201, + content_type="application/json", + ) + + result = repos.create_repo_gpg_key( + OWNER, + REPO, + gpg_private_key=FAKE_GPG_KEY_MATERIAL, + gpg_passphrase="s3cret", + ) + + assert result["fingerprint"] == body["fingerprint"] + + sent = _last_request() + sent_body = json.loads(sent.body) + assert sent_body["gpg_private_key"] == FAKE_GPG_KEY_MATERIAL + assert sent_body["gpg_passphrase"] == "s3cret" + + @httpretty.activate(allow_net_connect=False) + def test_omits_passphrase_when_none(self): + body = _gpg_key_body() + httpretty.register_uri( + httpretty.POST, + GPG_URL, + body=json.dumps(body), + status=201, + content_type="application/json", + ) + + repos.create_repo_gpg_key(OWNER, REPO, gpg_private_key=FAKE_GPG_KEY_MATERIAL) + + sent = _last_request() + sent_body = json.loads(sent.body) + assert "gpg_passphrase" not in sent_body + + @httpretty.activate(allow_net_connect=False) + def test_422_raises_api_exception_with_fields(self): + httpretty.register_uri( + httpretty.POST, + GPG_URL, + body=json.dumps( + { + "detail": "Invalid GPG key.", + "fields": {"gpg_private_key": ["Not a valid GPG private key."]}, + } + ), + status=422, + content_type="application/json", + ) + + with pytest.raises(ApiException) as exc_info: + repos.create_repo_gpg_key(OWNER, REPO, gpg_private_key="not-a-real-key") + + assert exc_info.value.status == 422 + assert "gpg_private_key" in exc_info.value.fields + + +class TestRegenerateRepoGpgKey: + @httpretty.activate(allow_net_connect=False) + def test_success_returns_new_gpg_key(self): + body = _gpg_key_body( + fingerprint="1111AAAA2222BBBB3333CCCC4444DDDD5555EEEE", + fingerprint_short="5555EEEE", + ) + httpretty.register_uri( + httpretty.POST, + GPG_REGENERATE_URL, + body=json.dumps(body), + status=200, + content_type="application/json", + ) + + result = repos.regenerate_repo_gpg_key(OWNER, REPO) + + assert result["fingerprint"] == body["fingerprint"] + assert result["fingerprint_short"] == body["fingerprint_short"] + + @httpretty.activate(allow_net_connect=False) + def test_404_raises_api_exception(self): + httpretty.register_uri( + httpretty.POST, + GPG_REGENERATE_URL, + body=json.dumps({"detail": "Not found."}), + status=404, + content_type="application/json", + ) + + with pytest.raises(ApiException) as exc_info: + repos.regenerate_repo_gpg_key(OWNER, REPO) + + assert exc_info.value.status == 404 From cb8c3c88cd1550d120cc55a777b2e2bbf871e54d Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Thu, 13 Aug 2026 09:23:55 +0100 Subject: [PATCH 2/8] fix(issue-83): harden repository GPG key upload --- cloudsmith_cli/cli/commands/repos.py | 53 ++-- .../cli/tests/commands/test_repos.py | 273 ++++++++++++++++-- 2 files changed, 280 insertions(+), 46 deletions(-) diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index a40e1cd7..91dc67a7 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -299,12 +299,8 @@ def print_gpg_key(gpg_key): @repositories.group(cls=command.AliasGroup, name="gpg", aliases=[]) -@decorators.common_cli_config_options -@decorators.common_cli_output_options -@decorators.common_api_auth_options -@decorators.initialise_api @click.pass_context -def gpg(ctx, opts): # pylint: disable=unused-argument +def gpg(ctx): # pylint: disable=unused-argument """ Manage a repository's GPG signing key. @@ -371,7 +367,8 @@ def gpg_get(ctx, opts, owner_repo): type=click.File("r"), required=True, help="Path to a file containing the armored GPG private key to upload. " - "Use '-' to read from stdin.", + "Use '-' to read from stdin; in that case --passphrase-file must be a " + "file path.", ) @click.option( "--passphrase-file", @@ -381,9 +378,10 @@ def gpg_get(ctx, opts, owner_repo): default=None, help="Path to a file containing the GPG private key's passphrase. If " "omitted, you'll be prompted for it interactively (leave blank if the " - "key has none). The passphrase is never accepted as a plain " - "command-line value, to avoid leaking it into shell history or the " - "process list.", + "key has none). One trailing line ending is ignored. Use '-' to read " + "from stdin only when --private-key-file is a path. The passphrase is " + "never accepted as a plain command-line value, to avoid leaking it " + "into shell history or the process list.", ) @click.pass_context def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): @@ -405,6 +403,22 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): owner, repo = owner_repo use_stderr = utils.should_use_stderr(opts) + if opts.debug: + raise click.BadParameter( + "Debug output is disabled for this command because the request " + "contains private key material and a passphrase.", + param_hint="--debug", + ) + + stdin = click.get_text_stream("stdin") + private_key_from_stdin = private_key_file is stdin + passphrase_from_stdin = passphrase_file is stdin + if private_key_from_stdin and (passphrase_file is None or passphrase_from_stdin): + raise click.BadParameter( + "Must be a file path (not '-') when --private-key-file is '-'.", + param_hint="--passphrase-file", + ) + gpg_private_key = private_key_file.read() if not gpg_private_key.strip(): raise click.BadParameter( @@ -412,17 +426,20 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): ) if passphrase_file is not None: - gpg_passphrase = passphrase_file.read().strip() or None + gpg_passphrase = passphrase_file.read() + if gpg_passphrase.endswith("\r\n"): + gpg_passphrase = gpg_passphrase[:-2] + elif gpg_passphrase.endswith(("\r", "\n")): + gpg_passphrase = gpg_passphrase[:-1] else: - gpg_passphrase = ( - click.prompt( - "GPG passphrase (leave blank if the key has none)", - hide_input=True, - default="", - show_default=False, - ).strip() - or None + gpg_passphrase = click.prompt( + "GPG passphrase (leave blank if the key has none)", + hide_input=True, + default="", + show_default=False, ) + if gpg_passphrase == "": + gpg_passphrase = None click.echo( f"Uploading GPG key for {click.style(repo, bold=True)} in the " diff --git a/cloudsmith_cli/cli/tests/commands/test_repos.py b/cloudsmith_cli/cli/tests/commands/test_repos.py index 95658b2d..efdda804 100644 --- a/cloudsmith_cli/cli/tests/commands/test_repos.py +++ b/cloudsmith_cli/cli/tests/commands/test_repos.py @@ -4,15 +4,8 @@ import pytest from ...commands.list_ import repos as list_repos -from ...commands.repos import ( - create, - delete, - get, - gpg_get, - gpg_regenerate, - gpg_upload, - update, -) +from ...commands.main import main +from ...commands.repos import create, delete, get, update from ..utils import random_str HERMETIC_ARGS = ["--api-key", "fake-api-key"] @@ -33,6 +26,18 @@ } +def gpg_command_args(command, *args): + """Build arguments that exercise the registered repos GPG command tree.""" + return [ + "repos", + *HERMETIC_ARGS, + "gpg", + command, + *args, + *HERMETIC_ARGS, + ] + + def create_repo_config_file(directory, name, description, repository_type_str, slug): """Create a REPO-CONFIG.json file in `directory` with the values provided.""" file_path = directory / "REPO_CONFIG.json" @@ -230,11 +235,16 @@ def test_repos_commands(runner, organization, tmp_path): class TestReposGpgGet: @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") - def test_success_prints_fingerprint(self, mock_list, runner): + @pytest.mark.parametrize("command_name", ["get", "list", "ls"]) + def test_registered_commands_print_fingerprint( + self, mock_list, runner, command_name + ): mock_list.return_value = dict(_GPG_KEY) result = runner.invoke( - gpg_get, ["my-org/my-repo", *HERMETIC_ARGS], catch_exceptions=False + main, + gpg_command_args(command_name, "my-org/my-repo"), + catch_exceptions=False, ) assert result.exit_code == 0, result.output @@ -247,8 +257,8 @@ def test_json_output(self, mock_list, runner): mock_list.return_value = dict(_GPG_KEY) result = runner.invoke( - gpg_get, - ["my-org/my-repo", "-F", "json", *HERMETIC_ARGS], + main, + gpg_command_args("get", "my-org/my-repo", "-F", "json"), catch_exceptions=False, ) @@ -263,12 +273,34 @@ def test_json_output(self, mock_list, runner): def test_invalid_owner_repo_rejected(self, runner): result = runner.invoke( - gpg_get, ["not-a-valid-argument", *HERMETIC_ARGS], catch_exceptions=False + main, + gpg_command_args("get", "not-a-valid-argument"), + catch_exceptions=False, ) assert result.exit_code != 0 assert "Must be in the form of OWNER/REPO" in result.output + def test_structural_group_does_not_advertise_inert_common_options(self, runner): + result = runner.invoke( + main, + ["repos", *HERMETIC_ARGS, "gpg", "--help"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "--output-format" not in result.output + assert "--debug" not in result.output + assert "--verbose" not in result.output + + misplaced = runner.invoke( + main, + ["repos", *HERMETIC_ARGS, "gpg", "-F", "json", "get", "my-org/my-repo"], + catch_exceptions=False, + ) + assert misplaced.exit_code != 0 + assert "No such option '-F'" in misplaced.output + class TestReposGpgUpload: @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") @@ -281,15 +313,15 @@ def test_uploads_key_and_passphrase_from_files(self, mock_create, runner, tmp_pa passphrase_file.write_text("s3cret\n") result = runner.invoke( - gpg_upload, - [ + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file), "--passphrase-file", str(passphrase_file), - *HERMETIC_ARGS, - ], + ), catch_exceptions=False, ) @@ -311,8 +343,10 @@ def test_prompts_for_passphrase_when_no_file_given( key_file.write_text(_FAKE_GPG_KEY_MATERIAL) result = runner.invoke( - gpg_upload, - ["my-org/my-repo", "--private-key-file", str(key_file), *HERMETIC_ARGS], + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file) + ), input="\n", catch_exceptions=False, ) @@ -325,14 +359,197 @@ def test_prompts_for_passphrase_when_no_file_given( gpg_passphrase=None, ) + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + @pytest.mark.parametrize( + "passphrase,expected", + [ + (" leading-space\n", " leading-space"), + ("trailing-space \n", "trailing-space "), + (" \t \n", " \t "), + ("windows-line-ending\r\n", "windows-line-ending"), + ], + ) + def test_passphrase_file_preserves_whitespace_and_removes_one_line_ending( + self, mock_create, runner, tmp_path, passphrase, expected + ): + mock_create.return_value = dict(_GPG_KEY) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + passphrase_file = tmp_path / "passphrase.txt" + passphrase_file.write_bytes(passphrase.encode()) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--passphrase-file", + str(passphrase_file), + ), + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert mock_create.call_args.kwargs["gpg_passphrase"] == expected + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + @pytest.mark.parametrize( + "passphrase", [" leading-space", "trailing-space ", " \t "] + ) + def test_prompt_preserves_passphrase_whitespace( + self, mock_create, runner, tmp_path, passphrase + ): + mock_create.return_value = dict(_GPG_KEY) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file) + ), + input=f"{passphrase}\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert mock_create.call_args.kwargs["gpg_passphrase"] == passphrase + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_private_key_stdin_requires_separate_passphrase_file( + self, mock_create, runner + ): + result = runner.invoke( + main, + gpg_command_args("upload", "my-org/my-repo", "--private-key-file", "-"), + input=_FAKE_GPG_KEY_MATERIAL, + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "--passphrase-file" in result.output + assert "Must be a file path (not '-')" in result.output + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_uploads_private_key_from_stdin_with_passphrase_file( + self, mock_create, runner, tmp_path + ): + mock_create.return_value = dict(_GPG_KEY) + passphrase_file = tmp_path / "passphrase.txt" + passphrase_file.write_text("passphrase\n") + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + "-", + "--passphrase-file", + str(passphrase_file), + ), + input=_FAKE_GPG_KEY_MATERIAL, + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + mock_create.assert_called_once_with( + "my-org", + "my-repo", + gpg_private_key=_FAKE_GPG_KEY_MATERIAL, + gpg_passphrase="passphrase", + ) + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_rejects_using_stdin_for_both_secret_inputs(self, mock_create, runner): + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + "-", + "--passphrase-file", + "-", + ), + input=_FAKE_GPG_KEY_MATERIAL, + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "Must be a file path (not '-')" in result.output + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_passphrase_can_be_read_from_stdin_when_key_uses_path( + self, mock_create, runner, tmp_path + ): + mock_create.return_value = dict(_GPG_KEY) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--passphrase-file", + "-", + ), + input=" stdin-passphrase \n", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert mock_create.call_args.kwargs["gpg_passphrase"] == " stdin-passphrase " + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_debug_is_rejected_without_disclosing_secrets( + self, mock_create, runner, tmp_path + ): + private_key = "private-key-debug-sentinel" + passphrase = "passphrase-debug-sentinel" + key_file = tmp_path / "key.asc" + key_file.write_text(private_key) + passphrase_file = tmp_path / "passphrase.txt" + passphrase_file.write_text(passphrase) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--passphrase-file", + str(passphrase_file), + "--debug", + ), + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "Debug output is disabled for this command" in result.output + for output in (result.stdout, result.stderr): + assert private_key not in output + assert passphrase not in output + mock_create.assert_not_called() + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") def test_empty_private_key_file_rejected(self, mock_create, runner, tmp_path): key_file = tmp_path / "key.asc" key_file.write_text(" \n") result = runner.invoke( - gpg_upload, - ["my-org/my-repo", "--private-key-file", str(key_file), *HERMETIC_ARGS], + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file) + ), catch_exceptions=False, ) @@ -343,8 +560,8 @@ def test_empty_private_key_file_rejected(self, mock_create, runner, tmp_path): def test_private_key_flag_not_accepted(self, runner): """Key material must never be a plain CLI value (shell-history/process-list leak).""" result = runner.invoke( - gpg_upload, - ["my-org/my-repo", "--private-key", "sekrit", *HERMETIC_ARGS], + main, + gpg_command_args("upload", "my-org/my-repo", "--private-key", "sekrit"), catch_exceptions=False, ) @@ -356,8 +573,8 @@ class TestReposGpgRegenerate: @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") def test_prompts_for_confirmation_and_declines(self, mock_regenerate, runner): result = runner.invoke( - gpg_regenerate, - ["my-org/my-repo", *HERMETIC_ARGS], + main, + gpg_command_args("regenerate", "my-org/my-repo"), input="N", catch_exceptions=False, ) @@ -372,8 +589,8 @@ def test_yes_flag_skips_confirmation(self, mock_regenerate, runner): mock_regenerate.return_value = new_key result = runner.invoke( - gpg_regenerate, - ["my-org/my-repo", "-y", *HERMETIC_ARGS], + main, + gpg_command_args("regenerate", "my-org/my-repo", "-y"), catch_exceptions=False, ) From df68bd419afbfe9e6c778af52be2930026466057 Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Thu, 13 Aug 2026 09:25:32 +0100 Subject: [PATCH 3/8] test(issue-83): cover GPG mutation JSON output --- .../cli/tests/commands/test_repos.py | 42 +++++++++++++++++++ 1 file changed, 42 insertions(+) diff --git a/cloudsmith_cli/cli/tests/commands/test_repos.py b/cloudsmith_cli/cli/tests/commands/test_repos.py index efdda804..03e46d15 100644 --- a/cloudsmith_cli/cli/tests/commands/test_repos.py +++ b/cloudsmith_cli/cli/tests/commands/test_repos.py @@ -333,6 +333,33 @@ def test_uploads_key_and_passphrase_from_files(self, mock_create, runner, tmp_pa gpg_passphrase="s3cret", ) + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_json_output(self, mock_create, runner, tmp_path): + mock_create.return_value = dict(_GPG_KEY) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + passphrase_file = tmp_path / "passphrase.txt" + passphrase_file.write_text("passphrase\n") + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--passphrase-file", + str(passphrase_file), + "-F", + "json", + ), + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + document = json.loads(result.stdout) + assert document["data"]["fingerprint"] == _GPG_KEY["fingerprint"] + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") def test_prompts_for_passphrase_when_no_file_given( self, mock_create, runner, tmp_path @@ -597,3 +624,18 @@ def test_yes_flag_skips_confirmation(self, mock_regenerate, runner): assert result.exit_code == 0, result.output mock_regenerate.assert_called_once_with("my-org", "my-repo") assert new_key["fingerprint"] in result.output + + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_pretty_json_output(self, mock_regenerate, runner): + mock_regenerate.return_value = dict(_GPG_KEY) + + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "-y", "-F", "pretty_json"), + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + document = json.loads(result.stdout) + assert document["data"]["fingerprint"] == _GPG_KEY["fingerprint"] + assert result.stdout.startswith("{\n ") From d03a122fb052d286df81e6ea322c1a133fd8f7ba Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Tue, 25 Aug 2026 15:58:33 +0100 Subject: [PATCH 4/8] test(issue-83): reset the thread-local CLI options between tests `config.get_or_create_options` caches the Options object in a thread-local, so state that is sticky by design - `--debug` in particular - leaked from one test's CLI invocation into every later one in the same process. Two GPG tests already failed because of it, and any test that runs after one passing `--debug` was at risk. The two tests that needed a clean Options object cleared the thread-local themselves; do it for every CLI test instead. --- cloudsmith_cli/cli/tests/conftest.py | 12 ++++++++++++ 1 file changed, 12 insertions(+) diff --git a/cloudsmith_cli/cli/tests/conftest.py b/cloudsmith_cli/cli/tests/conftest.py index 6ba3fe94..64b4f938 100644 --- a/cloudsmith_cli/cli/tests/conftest.py +++ b/cloudsmith_cli/cli/tests/conftest.py @@ -3,6 +3,7 @@ import click.testing import pytest +from ..config import OPTIONS from ...core.api.init import initialise_api from ...core.api.repos import create_repo, delete_repo from ...core.credentials.models import CredentialResult @@ -17,6 +18,17 @@ def _get_env_var_or_skip(key): return value +@pytest.fixture(autouse=True) +def reset_cli_options(monkeypatch): + """Discard the thread-local Options object between tests. + + ``config.get_or_create_options`` caches the Options object in a + thread-local, so state such as ``--debug`` (which is sticky by design) + would otherwise leak from one test's CLI invocation into the next. + """ + monkeypatch.delattr(OPTIONS, "value", raising=False) + + @pytest.fixture() def runner(): """Return a CliRunner with which to run Commands.""" From 8f55b09f92c3a89d64afe664d0a18f732b72d694 Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Tue, 25 Aug 2026 15:58:34 +0100 Subject: [PATCH 5/8] feat(issue-83): rework repos gpg confirmation, dry-run and error voice Applies the command-design review of `cloudsmith repos gpg`: - `regenerate` now requires the word "regenerate" to be typed, instead of a y/N answer, and states what is irrevocable about it before asking. Anything else typed declines with "Not confirmed. No changes made." and sends nothing. With no terminal attached the command fails with a usage error rather than blocking on a question nobody can answer, so `-y/--yes` stays the way to run it unattended. - Both mutating subcommands accept `-n/--dry-run`, which resolves and validates the inputs (so an empty key file still errors) and reports what would change without calling the API. - The GPG failures a person can act on now read as one sentence, e.g. "Could not set GPG key for your-repo: custom GPG keys require a paid plan." `handle_api_exceptions` takes an optional per-status summary map for this; unmapped statuses and JSON output keep the existing rendering, so nothing else changes shape. - `upload` only prompts for a passphrase when a terminal is attached; without one it takes the key to be unencrypted instead of aborting on EOF, which is what the documented behaviour always claimed. The prompt also goes to stderr under `-F json` so stdout stays a single parseable document. --- CHANGELOG.md | 4 + cloudsmith_cli/cli/commands/repos.py | 171 +++++++++- cloudsmith_cli/cli/exceptions.py | 104 +++--- .../cli/tests/commands/test_repos.py | 315 +++++++++++++++++- 4 files changed, 538 insertions(+), 56 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index c08c2ba0..23b13d9c 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 `cloudsmith repos gpg` for managing the GPG key a repository signs its package indexes with. `get` shows the active key and its armored public block, `upload` installs a key you supply, and `regenerate` replaces the current key with a freshly generated Cloudsmith one. Key material and passphrases are only ever read from a file, stdin, or a hidden prompt, never from a command-line value, and `--debug` is refused on `upload` so the request body can't be logged. Both mutating subcommands accept `-n/--dry-run`, and `regenerate` asks you to type `regenerate` to confirm - with no terminal attached it fails instead of blocking, so pass `-y/--yes` for unattended runs. There's no `delete` subcommand because the API has no way to remove a repository's key. + ## [1.25.0] - 2026-08-24 ### Added diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index 91dc67a7..54bb2a72 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -298,6 +298,69 @@ def print_gpg_key(gpg_key): click.echo() +#: Reasons for the API failures a person can actually act on, keyed by +#: status. Anything not listed here keeps the standard error rendering. +GPG_READ_ERROR_REASONS = {404: "not found"} +GPG_WRITE_ERROR_REASONS = { + 400: "the provided key is not valid", + 402: "custom GPG keys require a paid plan", + 404: "not found", +} + +REGENERATE_CONFIRMATION_WORD = "regenerate" + +REGENERATE_WARNING = ( + "Regenerating a repository's GPG key is irrevocable. The old key is discarded\n" + "and every consumer verifying against its fingerprint will need to fetch and\n" + "trust the new one before their next install succeeds." +) + + +def gpg_error_summaries(action, repo, reasons): + """Build single-line error messages for a GPG command, keyed by status.""" + return { + status: f"Could not {action} GPG key for {repo}: {reason}." + for status, reason in reasons.items() + } + + +def stdin_is_a_terminal(): + """Check whether stdin is attached to a terminal.""" + return click.get_text_stream("stdin").isatty() + + +def confirm_regenerate(err=False): + """Ask for typed confirmation before regenerating a repository's GPG key. + + Returns True only if the exact confirmation word was typed. Raises a + usage error when there's no terminal to ask, so an unattended run fails + fast instead of blocking on a question nobody can answer. + """ + if not stdin_is_a_terminal(): + raise click.UsageError( + "Refusing to regenerate the GPG key without confirmation: stdin is " + "not a terminal. Pass -y/--yes to confirm non-interactively." + ) + + click.echo(err=err) + click.echo(REGENERATE_WARNING, err=err) + click.echo(err=err) + + answer = click.prompt( + f"Type '{REGENERATE_CONFIRMATION_WORD}' to confirm", + default="", + show_default=False, + err=err, + ) + + if answer.strip() == REGENERATE_CONFIRMATION_WORD: + return True + + click.echo(err=err) + click.secho("Not confirmed. No changes made.", fg="yellow", err=err) + return False + + @repositories.group(cls=command.AliasGroup, name="gpg", aliases=[]) @click.pass_context def gpg(ctx): # pylint: disable=unused-argument @@ -340,7 +403,12 @@ def gpg_get(ctx, opts, owner_repo): context_msg = "Failed to get the repository GPG key!" with ( - handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + handle_api_exceptions( + ctx, + opts=opts, + context_msg=context_msg, + error_summaries=gpg_error_summaries("get", repo, GPG_READ_ERROR_REASONS), + ), maybe_spinner(opts), ): gpg_key = api.list_repo_gpg_key(owner, repo) @@ -383,8 +451,17 @@ def gpg_get(ctx, opts, owner_repo): "never accepted as a plain command-line value, to avoid leaking it " "into shell history or the process list.", ) +@click.option( + "-n", + "--dry-run", + "dry_run", + default=False, + is_flag=True, + help="Validate the inputs and show what would be uploaded, without " + "changing the repository's key.", +) @click.pass_context -def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): +def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run): """ Set (upload) the active GPG signing key for a repository. @@ -396,6 +473,9 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): The private key material is always read from a file (or stdin via '-'), never accepted as a plain command-line argument. + Use -n/--dry-run to validate the key and passphrase inputs without + changing the repository's key. + Full CLI example: $ cloudsmith repos gpg upload your-org/your-repo --private-key-file key.asc @@ -431,16 +511,44 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): gpg_passphrase = gpg_passphrase[:-2] elif gpg_passphrase.endswith(("\r", "\n")): gpg_passphrase = gpg_passphrase[:-1] - else: + elif stdin_is_a_terminal(): gpg_passphrase = click.prompt( "GPG passphrase (leave blank if the key has none)", hide_input=True, default="", show_default=False, + err=use_stderr, ) + else: + # Nothing to prompt: an unattended run would block on a question + # nobody can answer, so take the key to be unencrypted instead. + gpg_passphrase = "" if gpg_passphrase == "": gpg_passphrase = None + if dry_run: + if utils.maybe_print_as_json( + opts, + { + "dry_run": True, + "action": "upload", + "namespace": owner, + "repository": repo, + "passphrase_supplied": gpg_passphrase is not None, + }, + ): + return + + click.secho( + f"Would upload the GPG key for {click.style(repo, bold=True)} in the " + f"{click.style(owner, bold=True)} namespace " + f"({'with' if gpg_passphrase is not None else 'without'} a passphrase). " + "Nothing sent - this was a dry run.", + fg="yellow", + err=use_stderr, + ) + return + click.echo( f"Uploading GPG key for {click.style(repo, bold=True)} in the " f"{click.style(owner, bold=True)} namespace ... ", @@ -450,7 +558,12 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): context_msg = "Failed to set the repository GPG key!" with ( - handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + handle_api_exceptions( + ctx, + opts=opts, + context_msg=context_msg, + error_summaries=gpg_error_summaries("set", repo, GPG_WRITE_ERROR_REASONS), + ), maybe_spinner(opts), ): gpg_key = api.create_repo_gpg_key( @@ -480,8 +593,16 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file): is_flag=True, help="Assume yes as default answer to questions (this is dangerous!)", ) +@click.option( + "-n", + "--dry-run", + "dry_run", + default=False, + is_flag=True, + help="Show what would be regenerated, without changing the repository's key.", +) @click.pass_context -def gpg_regenerate(ctx, opts, owner_repo, yes): +def gpg_regenerate(ctx, opts, owner_repo, yes, dry_run): """ Regenerate the GPG signing key for a repository. @@ -494,6 +615,10 @@ def gpg_regenerate(ctx, opts, owner_repo, yes): one; consumers relying on the old key's fingerprint will need to pick up the new one. There is no way to undo this from the CLI. + Because of that, the command asks you to type 'regenerate' to confirm. + Unattended runs never block on that question: with no terminal attached + the command fails unless -y/--yes was passed. + Full CLI example: $ cloudsmith repos gpg regenerate your-org/your-repo @@ -501,18 +626,42 @@ def gpg_regenerate(ctx, opts, owner_repo, yes): owner, repo = owner_repo use_stderr = utils.should_use_stderr(opts) - prompt = ( - f"regenerate the GPG key for {click.style(repo, bold=True)} in the " - f"{click.style(owner, bold=True)} namespace" - ) - if not utils.confirm_operation(prompt, assume_yes=yes, err=use_stderr): + if dry_run: + if utils.maybe_print_as_json( + opts, + { + "dry_run": True, + "action": "regenerate", + "namespace": owner, + "repository": repo, + }, + ): + return + + click.secho( + f"Would regenerate the GPG key for {click.style(repo, bold=True)} in " + f"the {click.style(owner, bold=True)} namespace. Nothing sent - this " + "was a dry run.", + fg="yellow", + err=use_stderr, + ) + return + + if not yes and not confirm_regenerate(err=use_stderr): return click.echo("Regenerating GPG key ... ", nl=False, err=use_stderr) context_msg = "Failed to regenerate the repository GPG key!" with ( - handle_api_exceptions(ctx, opts=opts, context_msg=context_msg), + handle_api_exceptions( + ctx, + opts=opts, + context_msg=context_msg, + error_summaries=gpg_error_summaries( + "regenerate", repo, GPG_WRITE_ERROR_REASONS + ), + ), maybe_spinner(opts), ): gpg_key = api.regenerate_repo_gpg_key(owner, repo) diff --git a/cloudsmith_cli/cli/exceptions.py b/cloudsmith_cli/cli/exceptions.py index 7fe1b51e..daa28473 100644 --- a/cloudsmith_cli/cli/exceptions.py +++ b/cloudsmith_cli/cli/exceptions.py @@ -11,9 +11,21 @@ @contextlib.contextmanager def handle_api_exceptions( - ctx, opts, context_msg=None, nl=False, exit_on_error=True, reraise_on_error=False + ctx, + opts, + context_msg=None, + nl=False, + exit_on_error=True, + reraise_on_error=False, + error_summaries=None, ): - """Context manager that handles API exceptions.""" + """Context manager that handles API exceptions. + + ``error_summaries`` optionally maps an HTTP status to a single-line + message that replaces the default context/detail/hint block in + human-readable output. Statuses that aren't mapped (and JSON output) + keep the standard rendering. + """ # flake8: ignore=C901 # Use stderr for messages if the output is something else (e.g. # JSON) @@ -68,44 +80,14 @@ def handle_api_exceptions( else: click.secho("ERROR", fg="red", err=use_stderr) - click.secho( - f"{context_msg} (status: {exc.status} - {exc.status_description})", - fg="red", - err=use_stderr, - ) - - if detail or fields: - click.echo(err=use_stderr) - - if detail: - click.secho( - "Detail: {detail}".format( - detail=click.style(detail, fg="red", bold=False) - ), - bold=True, - err=use_stderr, - ) - - if fields: - for k, v in fields.items(): - field = f"{k.capitalize()} Field" - - # Flatten list/tuple error messages for text output - if isinstance(v, (list, tuple)): - v = " ".join(v) - - click.secho( - "{field}: {message}".format( - field=click.style(field, bold=True), - message=click.style(v, fg="red"), - ), - err=use_stderr, - ) - - if hint: - click.echo( - f"Hint: {click.style(hint, fg='yellow')}", - err=use_stderr, + summary = (error_summaries or {}).get(exc.status) + if summary: + # A command-specific one-liner replaces the generic + # context/detail/hint block for the statuses it covers. + click.secho(summary, fg="red", err=use_stderr) + else: + print_error_details( + context_msg, exc, detail, fields, hint, use_stderr=use_stderr ) if opts.verbose and not opts.debug and exc.headers: @@ -121,6 +103,48 @@ def handle_api_exceptions( ctx.exit(exc.status or 1) +def print_error_details(context_msg, exc, detail, fields, hint, use_stderr=False): + """Print the standard context/detail/fields/hint block for an error.""" + click.secho( + f"{context_msg} (status: {exc.status} - {exc.status_description})", + fg="red", + err=use_stderr, + ) + + if detail or fields: + click.echo(err=use_stderr) + + if detail: + click.secho( + "Detail: {detail}".format( + detail=click.style(detail, fg="red", bold=False) + ), + bold=True, + err=use_stderr, + ) + + for k, v in (fields or {}).items(): + field = f"{k.capitalize()} Field" + + # Flatten list/tuple error messages for text output + if isinstance(v, (list, tuple)): + v = " ".join(v) + + click.secho( + "{field}: {message}".format( + field=click.style(field, bold=True), + message=click.style(v, fg="red"), + ), + err=use_stderr, + ) + + if hint: + click.echo( + f"Hint: {click.style(hint, fg='yellow')}", + err=use_stderr, + ) + + def get_details(exc): """Get the details from the exception.""" detail = None diff --git a/cloudsmith_cli/cli/tests/commands/test_repos.py b/cloudsmith_cli/cli/tests/commands/test_repos.py index 03e46d15..88a0d920 100644 --- a/cloudsmith_cli/cli/tests/commands/test_repos.py +++ b/cloudsmith_cli/cli/tests/commands/test_repos.py @@ -3,6 +3,7 @@ import pytest +from ....core.api.exceptions import ApiException from ...commands.list_ import repos as list_repos from ...commands.main import main from ...commands.repos import create, delete, get, update @@ -360,9 +361,10 @@ def test_json_output(self, mock_create, runner, tmp_path): document = json.loads(result.stdout) assert document["data"]["fingerprint"] == _GPG_KEY["fingerprint"] + @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") def test_prompts_for_passphrase_when_no_file_given( - self, mock_create, runner, tmp_path + self, mock_create, _is_tty, runner, tmp_path ): mock_create.return_value = dict(_GPG_KEY) @@ -421,12 +423,13 @@ def test_passphrase_file_preserves_whitespace_and_removes_one_line_ending( assert result.exit_code == 0, result.output assert mock_create.call_args.kwargs["gpg_passphrase"] == expected + @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") @pytest.mark.parametrize( "passphrase", [" leading-space", "trailing-space ", " \t "] ) def test_prompt_preserves_passphrase_whitespace( - self, mock_create, runner, tmp_path, passphrase + self, mock_create, _is_tty, runner, tmp_path, passphrase ): mock_create.return_value = dict(_GPG_KEY) key_file = tmp_path / "key.asc" @@ -584,6 +587,143 @@ def test_empty_private_key_file_rejected(self, mock_create, runner, tmp_path): assert "private key file is empty" in result.output mock_create.assert_not_called() + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_no_terminal_means_no_passphrase_rather_than_a_stall( + self, mock_create, runner, tmp_path + ): + """An unattended run uploads an unencrypted key instead of blocking.""" + mock_create.return_value = dict(_GPG_KEY) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file) + ), + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "GPG passphrase" not in result.output + mock_create.assert_called_once_with( + "my-org", + "my-repo", + gpg_private_key=_FAKE_GPG_KEY_MATERIAL, + gpg_passphrase=None, + ) + + @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_passphrase_prompt_keeps_out_of_json_output( + self, mock_create, _is_tty, runner, tmp_path + ): + """stdout must stay a single parseable document when JSON is asked for.""" + mock_create.return_value = dict(_GPG_KEY) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "-F", + "json", + ), + input="\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + document = json.loads(result.stdout) + assert document["data"]["fingerprint"] == _GPG_KEY["fingerprint"] + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_dry_run_validates_inputs_but_sends_nothing( + self, mock_create, runner, tmp_path + ): + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + passphrase_file = tmp_path / "pass.txt" + passphrase_file.write_text("hunter2\n") + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--passphrase-file", + str(passphrase_file), + "--dry-run", + ), + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "Would upload the GPG key" in result.output + assert "with a passphrase" in result.output + assert "hunter2" not in result.output + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_dry_run_still_rejects_an_empty_key_file( + self, mock_create, runner, tmp_path + ): + key_file = tmp_path / "key.asc" + key_file.write_text(" \n") + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--dry-run", + ), + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "private key file is empty" in result.output + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_dry_run_json_output(self, mock_create, runner, tmp_path): + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--dry-run", + "-F", + "json", + ), + input="\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + document = json.loads(result.stdout) + assert document["data"] == { + "dry_run": True, + "action": "upload", + "namespace": "my-org", + "repository": "my-repo", + "passphrase_supplied": False, + } + mock_create.assert_not_called() + def test_private_key_flag_not_accepted(self, runner): """Key material must never be a plain CLI value (shell-history/process-list leak).""" result = runner.invoke( @@ -597,17 +737,84 @@ def test_private_key_flag_not_accepted(self, runner): class TestReposGpgRegenerate: + @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_typed_confirmation_regenerates(self, mock_regenerate, _is_tty, runner): + new_key = dict(_GPG_KEY, fingerprint="9999FFFF8888EEEE7777DDDD6666CCCC5555BBBB") + mock_regenerate.return_value = new_key + + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo"), + input="regenerate\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "irrevocable" in result.output + assert "Type 'regenerate' to confirm" in result.output + mock_regenerate.assert_called_once_with("my-org", "my-repo") + assert new_key["fingerprint"] in result.output + + @pytest.mark.parametrize("answer", ["", "n", "REGENERATE", "regen"]) + @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_anything_but_the_word_declines( + self, mock_regenerate, _is_tty, runner, answer + ): + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo"), + input=f"{answer}\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "Not confirmed. No changes made." in result.output + mock_regenerate.assert_not_called() + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") - def test_prompts_for_confirmation_and_declines(self, mock_regenerate, runner): + def test_fails_fast_without_a_terminal(self, mock_regenerate, runner): + """An unattended run must fail, not block on a question nobody can answer.""" result = runner.invoke( main, gpg_command_args("regenerate", "my-org/my-repo"), - input="N", + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "stdin is not a terminal" in result.output + assert "-y/--yes" in result.output + mock_regenerate.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_dry_run_sends_nothing(self, mock_regenerate, runner): + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "--dry-run"), catch_exceptions=False, ) assert result.exit_code == 0, result.output - assert "OK, phew! Close call. :-)" in result.output + assert "Would regenerate the GPG key" in result.output + mock_regenerate.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_dry_run_json_output(self, mock_regenerate, runner): + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "--dry-run", "-F", "json"), + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + document = json.loads(result.stdout) + assert document["data"] == { + "dry_run": True, + "action": "regenerate", + "namespace": "my-org", + "repository": "my-repo", + } mock_regenerate.assert_not_called() @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") @@ -639,3 +846,101 @@ def test_pretty_json_output(self, mock_regenerate, runner): document = json.loads(result.stdout) assert document["data"]["fingerprint"] == _GPG_KEY["fingerprint"] assert result.stdout.startswith("{\n ") + + +class TestReposGpgErrorVoice: + """The failures a person can act on read as one plain sentence.""" + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + def test_get_not_found(self, mock_list, runner): + mock_list.side_effect = ApiException(status=404, detail="Not found.") + + result = runner.invoke( + main, + gpg_command_args("get", "my-org/my-repo"), + catch_exceptions=False, + ) + + # AliasGroup.main runs click with standalone_mode=False, so the status + # comes back as the command's return value rather than an exit code. + assert result.return_value == 404 + assert "Could not get GPG key for my-repo: not found." in result.output + assert "status: 404" not in result.output + + @pytest.mark.parametrize( + ("status", "expected"), + [ + (400, "Could not set GPG key for my-repo: the provided key is not valid."), + ( + 402, + "Could not set GPG key for my-repo: custom GPG keys require a paid " + "plan.", + ), + (404, "Could not set GPG key for my-repo: not found."), + ], + ) + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_upload_failures(self, mock_create, runner, tmp_path, status, expected): + mock_create.side_effect = ApiException(status=status, detail="Whatever.") + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file) + ), + input="\n", + catch_exceptions=False, + ) + + assert result.return_value == status + assert expected in result.output + assert "Whatever." not in result.output + + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_regenerate_failure(self, mock_regenerate, runner): + mock_regenerate.side_effect = ApiException( + status=402, detail="Custom GPG keys are not active; upgrade your account!" + ) + + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "-y"), + catch_exceptions=False, + ) + + assert result.return_value == 402 + assert ( + "Could not regenerate GPG key for my-repo: custom GPG keys require a " + "paid plan." in result.output + ) + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + def test_unmapped_status_keeps_the_standard_rendering(self, mock_list, runner): + mock_list.side_effect = ApiException(status=500, detail="Boom.") + + result = runner.invoke( + main, + gpg_command_args("get", "my-org/my-repo"), + catch_exceptions=False, + ) + + assert result.return_value == 500 + assert "Failed to get the repository GPG key!" in result.output + assert "Detail: Boom." in result.output + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + def test_json_output_keeps_the_full_error_envelope(self, mock_list, runner): + mock_list.side_effect = ApiException(status=404, detail="Not found.") + + result = runner.invoke( + main, + gpg_command_args("get", "my-org/my-repo", "-F", "json"), + catch_exceptions=False, + ) + + assert result.return_value == 404 + document = json.loads(result.stdout) + assert document["detail"] == "Not found." + assert document["meta"]["code"] == 404 From e9516de24021ff129e2d38a7e3294c15530ee137 Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Tue, 25 Aug 2026 16:08:58 +0100 Subject: [PATCH 6/8] fix(issue-83): detect stdin secret inputs by their raw value The `gpg upload` guards that stop the private key and the passphrase both being read from stdin compared the streams click returned against `click.get_text_stream("stdin")`. `click.File` builds a fresh `_NonClosingTextIOWrapper` for `-` on every conversion, so that identity only ever holds under `CliRunner`, where the runner's own stdin object is handed back unchanged. In a real process both guards were inert: `--private-key-file - --passphrase-file -` read the key and then took the passphrase from an already-drained stdin, and `--private-key-file -` on its own silently assumed the key was unencrypted. Record the literal value each secret file option was given on the click context instead, and key the guards off that. This is exact in both a real process and under `CliRunner`, so the existing tests now prove the behaviour they claim to. Also sort the conftest imports, which the new `OPTIONS` import left out of order. --- cloudsmith_cli/cli/commands/repos.py | 36 ++++++++++++++++++++++++---- cloudsmith_cli/cli/tests/conftest.py | 2 +- 2 files changed, 32 insertions(+), 6 deletions(-) diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index 54bb2a72..353acaf3 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -316,6 +316,33 @@ def print_gpg_key(gpg_key): ) +class SecretFile(click.File): + """A ``click.File`` that also records the literal value it was given. + + ``click.File`` builds a brand new stream object for ``-`` every time it + converts a value, so the stdin-conflict checks in ``gpg upload`` cannot + be made by comparing the returned stream against + ``click.get_text_stream("stdin")``: that identity happens to hold under + ``CliRunner`` but never holds in a real process, which would leave the + checks silently inert exactly where they matter. Recording the raw + value instead makes them exact in both. + """ + + #: Key under which the raw values are stashed on the click context. + META_KEY = "cloudsmith_cli.secret_file_sources" + + def convert(self, value, param, ctx): + stream = super().convert(value, param, ctx) + if ctx is not None and param is not None and isinstance(value, str): + ctx.meta.setdefault(self.META_KEY, {})[param.name] = value + return stream + + +def secret_file_is_stdin(ctx, param_name): + """Check whether a secret file parameter was given as '-' (i.e. stdin).""" + return ctx.meta.get(SecretFile.META_KEY, {}).get(param_name) == "-" + + def gpg_error_summaries(action, repo, reasons): """Build single-line error messages for a GPG command, keyed by status.""" return { @@ -432,7 +459,7 @@ def gpg_get(ctx, opts, owner_repo): @click.option( "--private-key-file", "private_key_file", - type=click.File("r"), + type=SecretFile("r"), required=True, help="Path to a file containing the armored GPG private key to upload. " "Use '-' to read from stdin; in that case --passphrase-file must be a " @@ -441,7 +468,7 @@ def gpg_get(ctx, opts, owner_repo): @click.option( "--passphrase-file", "passphrase_file", - type=click.File("r"), + type=SecretFile("r"), required=False, default=None, help="Path to a file containing the GPG private key's passphrase. If " @@ -490,9 +517,8 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run param_hint="--debug", ) - stdin = click.get_text_stream("stdin") - private_key_from_stdin = private_key_file is stdin - passphrase_from_stdin = passphrase_file is stdin + private_key_from_stdin = secret_file_is_stdin(ctx, "private_key_file") + passphrase_from_stdin = secret_file_is_stdin(ctx, "passphrase_file") if private_key_from_stdin and (passphrase_file is None or passphrase_from_stdin): raise click.BadParameter( "Must be a file path (not '-') when --private-key-file is '-'.", diff --git a/cloudsmith_cli/cli/tests/conftest.py b/cloudsmith_cli/cli/tests/conftest.py index 64b4f938..8e139130 100644 --- a/cloudsmith_cli/cli/tests/conftest.py +++ b/cloudsmith_cli/cli/tests/conftest.py @@ -3,10 +3,10 @@ import click.testing import pytest -from ..config import OPTIONS from ...core.api.init import initialise_api from ...core.api.repos import create_repo, delete_repo from ...core.credentials.models import CredentialResult +from ..config import OPTIONS from .utils import random_str From ec875e3f5807891d52e56d1c26e4fb639678da32 Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Tue, 25 Aug 2026 16:13:22 +0100 Subject: [PATCH 7/8] feat(issue-83): make the gpg dry runs check the key they would replace A dry run that only echoed its own arguments could not catch the two mistakes it is there to catch: a mistyped repository and a credential that no longer works. Both mutating subcommands now read the key currently in place first, report the fingerprint the real run would replace, and stop before the mutating request - so those failures land in the rehearsal, in the same voice the real command would use. `upload --dry-run` also no longer prompts for the passphrase. Nothing is being sent, so there is no reason to make anyone type a real secret; the line says which source the real run would use instead. --- CHANGELOG.md | 2 +- cloudsmith_cli/cli/commands/repos.py | 109 ++++++++++++------ .../cli/tests/commands/test_repos.py | 107 +++++++++++++++-- 3 files changed, 170 insertions(+), 48 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 23b13d9c..3fa65a85 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,7 +9,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added -- Added `cloudsmith repos gpg` for managing the GPG key a repository signs its package indexes with. `get` shows the active key and its armored public block, `upload` installs a key you supply, and `regenerate` replaces the current key with a freshly generated Cloudsmith one. Key material and passphrases are only ever read from a file, stdin, or a hidden prompt, never from a command-line value, and `--debug` is refused on `upload` so the request body can't be logged. Both mutating subcommands accept `-n/--dry-run`, and `regenerate` asks you to type `regenerate` to confirm - with no terminal attached it fails instead of blocking, so pass `-y/--yes` for unattended runs. There's no `delete` subcommand because the API has no way to remove a repository's key. +- Added `cloudsmith repos gpg` for managing the GPG key a repository signs its package indexes with. `get` shows the active key and its armored public block, `upload` installs a key you supply, and `regenerate` replaces the current key with a freshly generated Cloudsmith one. Key material and passphrases are only ever read from a file, stdin, or a hidden prompt, never from a command-line value, and `--debug` is refused on `upload` so the request body can't be logged. Both mutating subcommands accept `-n/--dry-run`, which checks the inputs and the key currently in place - naming the fingerprint that would be replaced - then stops before the request, so a mistyped repository or a stale credential fails there rather than on the real attempt. `regenerate` asks you to type `regenerate` to confirm - with no terminal attached it fails instead of blocking, so pass `-y/--yes` for unattended runs. There's no `delete` subcommand because the API has no way to remove a repository's key. ## [1.25.0] - 2026-08-24 diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index 353acaf3..cb251735 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -351,6 +351,48 @@ def gpg_error_summaries(action, repo, reasons): } +def gpg_dry_run(ctx, opts, action, reasons, owner, repo, note=None, err=False): + """Report what a mutating GPG command would do, without doing it. + + Reads the key that's in place first: that proves the credential can + reach the repository and names the fingerprint the real run would + replace, so a mistyped repository or an expired token fails here rather + than on the attempt itself. + """ + context_msg = f"Failed to {action} the repository GPG key!" + with ( + handle_api_exceptions( + ctx, + opts=opts, + context_msg=context_msg, + error_summaries=gpg_error_summaries(action, repo, reasons), + ), + maybe_spinner(opts), + ): + current = api.list_repo_gpg_key(owner, repo) + + if utils.maybe_print_as_json( + opts, + { + "dry_run": True, + "action": action, + "namespace": owner, + "repository": repo, + "current_fingerprint": current.get("fingerprint"), + }, + ): + return + + click.secho( + f"Would {action} the GPG key for {click.style(repo, bold=True)} in the " + f"{click.style(owner, bold=True)} namespace, replacing " + f"{current.get('fingerprint') or '(none)'}" + f"{f' ({note})' if note else ''}. Nothing sent - this was a dry run.", + fg="yellow", + err=err, + ) + + def stdin_is_a_terminal(): """Check whether stdin is attached to a terminal.""" return click.get_text_stream("stdin").isatty() @@ -484,8 +526,8 @@ def gpg_get(ctx, opts, owner_repo): "dry_run", default=False, is_flag=True, - help="Validate the inputs and show what would be uploaded, without " - "changing the repository's key.", + help="Check the inputs and the key currently in place, and show what " + "would be uploaded, without changing the repository's key.", ) @click.pass_context def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run): @@ -500,8 +542,8 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run The private key material is always read from a file (or stdin via '-'), never accepted as a plain command-line argument. - Use -n/--dry-run to validate the key and passphrase inputs without - changing the repository's key. + Use -n/--dry-run to check the key and passphrase inputs, and the key + currently in place, without changing anything. Full CLI example: @@ -537,6 +579,16 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run gpg_passphrase = gpg_passphrase[:-2] elif gpg_passphrase.endswith(("\r", "\n")): gpg_passphrase = gpg_passphrase[:-1] + passphrase_note = "with the passphrase from --passphrase-file" + elif dry_run: + # Nothing is being sent, so there's no reason to make anyone type a + # real secret. Say which source the real run would use instead. + gpg_passphrase = "" + passphrase_note = ( + "you'd be asked for the passphrase" + if stdin_is_a_terminal() + else "with no passphrase" + ) elif stdin_is_a_terminal(): gpg_passphrase = click.prompt( "GPG passphrase (leave blank if the key has none)", @@ -545,32 +597,24 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run show_default=False, err=use_stderr, ) + passphrase_note = "with the passphrase you were asked for" else: # Nothing to prompt: an unattended run would block on a question # nobody can answer, so take the key to be unencrypted instead. gpg_passphrase = "" + passphrase_note = "with no passphrase" if gpg_passphrase == "": gpg_passphrase = None if dry_run: - if utils.maybe_print_as_json( + gpg_dry_run( + ctx, opts, - { - "dry_run": True, - "action": "upload", - "namespace": owner, - "repository": repo, - "passphrase_supplied": gpg_passphrase is not None, - }, - ): - return - - click.secho( - f"Would upload the GPG key for {click.style(repo, bold=True)} in the " - f"{click.style(owner, bold=True)} namespace " - f"({'with' if gpg_passphrase is not None else 'without'} a passphrase). " - "Nothing sent - this was a dry run.", - fg="yellow", + "set", + GPG_WRITE_ERROR_REASONS, + owner, + repo, + note=passphrase_note, err=use_stderr, ) return @@ -625,7 +669,7 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run "dry_run", default=False, is_flag=True, - help="Show what would be regenerated, without changing the repository's key.", + help="Show which key would be replaced, without changing the repository's key.", ) @click.pass_context def gpg_regenerate(ctx, opts, owner_repo, yes, dry_run): @@ -653,22 +697,13 @@ def gpg_regenerate(ctx, opts, owner_repo, yes, dry_run): use_stderr = utils.should_use_stderr(opts) if dry_run: - if utils.maybe_print_as_json( + gpg_dry_run( + ctx, opts, - { - "dry_run": True, - "action": "regenerate", - "namespace": owner, - "repository": repo, - }, - ): - return - - click.secho( - f"Would regenerate the GPG key for {click.style(repo, bold=True)} in " - f"the {click.style(owner, bold=True)} namespace. Nothing sent - this " - "was a dry run.", - fg="yellow", + "regenerate", + GPG_WRITE_ERROR_REASONS, + owner, + repo, err=use_stderr, ) return diff --git a/cloudsmith_cli/cli/tests/commands/test_repos.py b/cloudsmith_cli/cli/tests/commands/test_repos.py index 88a0d920..7d958e30 100644 --- a/cloudsmith_cli/cli/tests/commands/test_repos.py +++ b/cloudsmith_cli/cli/tests/commands/test_repos.py @@ -641,10 +641,12 @@ def test_passphrase_prompt_keeps_out_of_json_output( document = json.loads(result.stdout) assert document["data"]["fingerprint"] == _GPG_KEY["fingerprint"] + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") - def test_dry_run_validates_inputs_but_sends_nothing( - self, mock_create, runner, tmp_path + def test_dry_run_names_the_key_it_would_replace( + self, mock_create, mock_list, runner, tmp_path ): + mock_list.return_value = dict(_GPG_KEY) key_file = tmp_path / "key.asc" key_file.write_text(_FAKE_GPG_KEY_MATERIAL) passphrase_file = tmp_path / "pass.txt" @@ -665,9 +667,65 @@ def test_dry_run_validates_inputs_but_sends_nothing( ) assert result.exit_code == 0, result.output - assert "Would upload the GPG key" in result.output - assert "with a passphrase" in result.output + assert "Would set the GPG key" in result.output + assert _GPG_KEY["fingerprint"] in result.output + assert "--passphrase-file" in result.output assert "hunter2" not in result.output + mock_list.assert_called_once_with("my-org", "my-repo") + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_dry_run_reports_an_unreachable_repository( + self, mock_create, mock_list, runner, tmp_path + ): + """A mistyped repository or a stale token fails here, not on the real run.""" + mock_list.side_effect = ApiException(status=404, detail="Not found.") + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--dry-run", + ), + catch_exceptions=False, + ) + + assert result.return_value == 404 + assert "Could not set GPG key for my-repo: not found." in result.output + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_dry_run_never_asks_for_a_passphrase( + self, mock_create, mock_list, _is_tty, runner, tmp_path + ): + """Nothing is sent, so nobody should have to type a real secret.""" + mock_list.return_value = dict(_GPG_KEY) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--dry-run", + ), + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "GPG passphrase" not in result.output + assert "you'd be asked for the passphrase" in result.output mock_create.assert_not_called() @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") @@ -693,8 +751,10 @@ def test_dry_run_still_rejects_an_empty_key_file( assert "private key file is empty" in result.output mock_create.assert_not_called() + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") - def test_dry_run_json_output(self, mock_create, runner, tmp_path): + def test_dry_run_json_output(self, mock_create, mock_list, runner, tmp_path): + mock_list.return_value = dict(_GPG_KEY) key_file = tmp_path / "key.asc" key_file.write_text(_FAKE_GPG_KEY_MATERIAL) @@ -709,7 +769,6 @@ def test_dry_run_json_output(self, mock_create, runner, tmp_path): "-F", "json", ), - input="\n", catch_exceptions=False, ) @@ -717,10 +776,10 @@ def test_dry_run_json_output(self, mock_create, runner, tmp_path): document = json.loads(result.stdout) assert document["data"] == { "dry_run": True, - "action": "upload", + "action": "set", "namespace": "my-org", "repository": "my-repo", - "passphrase_supplied": False, + "current_fingerprint": _GPG_KEY["fingerprint"], } mock_create.assert_not_called() @@ -787,8 +846,13 @@ def test_fails_fast_without_a_terminal(self, mock_regenerate, runner): assert "-y/--yes" in result.output mock_regenerate.assert_not_called() + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") - def test_dry_run_sends_nothing(self, mock_regenerate, runner): + def test_dry_run_names_the_key_it_would_replace( + self, mock_regenerate, mock_list, runner + ): + mock_list.return_value = dict(_GPG_KEY) + result = runner.invoke( main, gpg_command_args("regenerate", "my-org/my-repo", "--dry-run"), @@ -797,10 +861,32 @@ def test_dry_run_sends_nothing(self, mock_regenerate, runner): assert result.exit_code == 0, result.output assert "Would regenerate the GPG key" in result.output + assert _GPG_KEY["fingerprint"] in result.output + mock_list.assert_called_once_with("my-org", "my-repo") mock_regenerate.assert_not_called() + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") - def test_dry_run_json_output(self, mock_regenerate, runner): + def test_dry_run_reports_an_unreachable_repository( + self, mock_regenerate, mock_list, runner + ): + mock_list.side_effect = ApiException(status=404, detail="Not found.") + + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "--dry-run"), + catch_exceptions=False, + ) + + assert result.return_value == 404 + assert "Could not regenerate GPG key for my-repo: not found." in result.output + mock_regenerate.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_dry_run_json_output(self, mock_regenerate, mock_list, runner): + mock_list.return_value = dict(_GPG_KEY) + result = runner.invoke( main, gpg_command_args("regenerate", "my-org/my-repo", "--dry-run", "-F", "json"), @@ -814,6 +900,7 @@ def test_dry_run_json_output(self, mock_regenerate, runner): "action": "regenerate", "namespace": "my-org", "repository": "my-repo", + "current_fingerprint": _GPG_KEY["fingerprint"], } mock_regenerate.assert_not_called() From c0b030b3031bbcc28e7a2cf176b6be6a140053b5 Mon Sep 17 00:00:00 2001 From: Bartosz Blizniak Date: Wed, 26 Aug 2026 13:56:19 +0100 Subject: [PATCH 8/8] fix(issue-83): address gpg command review feedback Binary (non-armored) --private-key-file now fails with a friendly message instead of a raw UnicodeDecodeError. Error summaries for GPG failures now name owner/repo instead of just repo, since a 404 can be caused by either. The 400 "provided key is not valid" mapping is scoped to the real upload request only - regenerate and both dry-run pre-flights never send a key, so they fall through to the standard rendering on a 400 instead of misattributing it. The dry-run pre-flight now prints progress text before its read, so a failure appends ERROR to that line instead of printing bare. --- cloudsmith_cli/cli/commands/repos.py | 42 ++++-- .../cli/tests/commands/test_repos.py | 134 ++++++++++++++++-- 2 files changed, 157 insertions(+), 19 deletions(-) diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index cb251735..76dfdaa2 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -301,11 +301,18 @@ def print_gpg_key(gpg_key): #: Reasons for the API failures a person can actually act on, keyed by #: status. Anything not listed here keeps the standard error rendering. GPG_READ_ERROR_REASONS = {404: "not found"} -GPG_WRITE_ERROR_REASONS = { +GPG_UPLOAD_ERROR_REASONS = { 400: "the provided key is not valid", 402: "custom GPG keys require a paid plan", 404: "not found", } +#: Statuses for GPG flows that never send a key in the request - regenerate, +#: and the read-only pre-flight both dry runs do - so a 400 can't plausibly +#: mean "the provided key is invalid" the way it does for an actual upload. +GPG_NO_KEY_ERROR_REASONS = { + 402: "custom GPG keys require a paid plan", + 404: "not found", +} REGENERATE_CONFIRMATION_WORD = "regenerate" @@ -343,10 +350,10 @@ def secret_file_is_stdin(ctx, param_name): return ctx.meta.get(SecretFile.META_KEY, {}).get(param_name) == "-" -def gpg_error_summaries(action, repo, reasons): +def gpg_error_summaries(action, owner, repo, reasons): """Build single-line error messages for a GPG command, keyed by status.""" return { - status: f"Could not {action} GPG key for {repo}: {reason}." + status: f"Could not {action} GPG key for {owner}/{repo}: {reason}." for status, reason in reasons.items() } @@ -359,18 +366,22 @@ def gpg_dry_run(ctx, opts, action, reasons, owner, repo, note=None, err=False): replace, so a mistyped repository or an expired token fails here rather than on the attempt itself. """ + click.echo("Checking current GPG key ... ", nl=False, err=err) + context_msg = f"Failed to {action} the repository GPG key!" with ( handle_api_exceptions( ctx, opts=opts, context_msg=context_msg, - error_summaries=gpg_error_summaries(action, repo, reasons), + error_summaries=gpg_error_summaries(action, owner, repo, reasons), ), maybe_spinner(opts), ): current = api.list_repo_gpg_key(owner, repo) + click.secho("OK", fg="green", err=err) + if utils.maybe_print_as_json( opts, { @@ -476,7 +487,9 @@ def gpg_get(ctx, opts, owner_repo): ctx, opts=opts, context_msg=context_msg, - error_summaries=gpg_error_summaries("get", repo, GPG_READ_ERROR_REASONS), + error_summaries=gpg_error_summaries( + "get", owner, repo, GPG_READ_ERROR_REASONS + ), ), maybe_spinner(opts), ): @@ -567,7 +580,14 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run param_hint="--passphrase-file", ) - gpg_private_key = private_key_file.read() + try: + gpg_private_key = private_key_file.read() + except UnicodeDecodeError as exc: + raise click.BadParameter( + "This looks like a binary key export, not an armored (text) one. " + "Re-export with 'gpg --armor --export-secret-keys' and try again.", + param_hint="--private-key-file", + ) from exc if not gpg_private_key.strip(): raise click.BadParameter( "The private key file is empty.", param_hint="--private-key-file" @@ -611,7 +631,7 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run ctx, opts, "set", - GPG_WRITE_ERROR_REASONS, + GPG_NO_KEY_ERROR_REASONS, owner, repo, note=passphrase_note, @@ -632,7 +652,9 @@ def gpg_upload(ctx, opts, owner_repo, private_key_file, passphrase_file, dry_run ctx, opts=opts, context_msg=context_msg, - error_summaries=gpg_error_summaries("set", repo, GPG_WRITE_ERROR_REASONS), + error_summaries=gpg_error_summaries( + "set", owner, repo, GPG_UPLOAD_ERROR_REASONS + ), ), maybe_spinner(opts), ): @@ -701,7 +723,7 @@ def gpg_regenerate(ctx, opts, owner_repo, yes, dry_run): ctx, opts, "regenerate", - GPG_WRITE_ERROR_REASONS, + GPG_NO_KEY_ERROR_REASONS, owner, repo, err=use_stderr, @@ -720,7 +742,7 @@ def gpg_regenerate(ctx, opts, owner_repo, yes, dry_run): opts=opts, context_msg=context_msg, error_summaries=gpg_error_summaries( - "regenerate", repo, GPG_WRITE_ERROR_REASONS + "regenerate", owner, repo, GPG_NO_KEY_ERROR_REASONS ), ), maybe_spinner(opts), diff --git a/cloudsmith_cli/cli/tests/commands/test_repos.py b/cloudsmith_cli/cli/tests/commands/test_repos.py index 7d958e30..74f32fb9 100644 --- a/cloudsmith_cli/cli/tests/commands/test_repos.py +++ b/cloudsmith_cli/cli/tests/commands/test_repos.py @@ -667,6 +667,7 @@ def test_dry_run_names_the_key_it_would_replace( ) assert result.exit_code == 0, result.output + assert "Checking current GPG key ... OK" in result.output assert "Would set the GPG key" in result.output assert _GPG_KEY["fingerprint"] in result.output assert "--passphrase-file" in result.output @@ -697,7 +698,37 @@ def test_dry_run_reports_an_unreachable_repository( ) assert result.return_value == 404 - assert "Could not set GPG key for my-repo: not found." in result.output + assert "Checking current GPG key ... ERROR" in result.output + assert "Could not set GPG key for my-org/my-repo: not found." in result.output + mock_create.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_dry_run_400_keeps_the_standard_rendering( + self, mock_create, mock_list, runner, tmp_path + ): + """The dry-run pre-flight is a GET; a 400 there can't mean "bad key" either.""" + mock_list.side_effect = ApiException( + status=400, detail="Some other validation problem." + ) + key_file = tmp_path / "key.asc" + key_file.write_text(_FAKE_GPG_KEY_MATERIAL) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--dry-run", + ), + catch_exceptions=False, + ) + + assert result.return_value == 400 + assert "the provided key is not valid" not in result.output + assert "Detail: Some other validation problem." in result.output mock_create.assert_not_called() @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) @@ -794,6 +825,26 @@ def test_private_key_flag_not_accepted(self, runner): assert result.exit_code != 0 assert "no such option" in result.output.lower() + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + def test_binary_key_file_fails_cleanly(self, mock_create, runner, tmp_path): + """A non-armored (binary) export must not surface a raw UnicodeDecodeError.""" + key_file = tmp_path / "key.gpg" + key_file.write_bytes(bytes(range(256))) + + result = runner.invoke( + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file) + ), + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "UnicodeDecodeError" not in result.output + assert "binary key export" in result.output + assert "--armor" in result.output + mock_create.assert_not_called() + class TestReposGpgRegenerate: @patch("cloudsmith_cli.cli.commands.repos.stdin_is_a_terminal", return_value=True) @@ -860,6 +911,7 @@ def test_dry_run_names_the_key_it_would_replace( ) assert result.exit_code == 0, result.output + assert "Checking current GPG key ... OK" in result.output assert "Would regenerate the GPG key" in result.output assert _GPG_KEY["fingerprint"] in result.output mock_list.assert_called_once_with("my-org", "my-repo") @@ -879,7 +931,31 @@ def test_dry_run_reports_an_unreachable_repository( ) assert result.return_value == 404 - assert "Could not regenerate GPG key for my-repo: not found." in result.output + assert "Checking current GPG key ... ERROR" in result.output + assert ( + "Could not regenerate GPG key for my-org/my-repo: not found." + in result.output + ) + mock_regenerate.assert_not_called() + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_dry_run_json_output_stays_parseable_on_error( + self, mock_regenerate, mock_list, runner + ): + """The new pre-flight progress text must not land on stdout with -F json.""" + mock_list.side_effect = ApiException(status=404, detail="Not found.") + + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "--dry-run", "-F", "json"), + catch_exceptions=False, + ) + + assert result.return_value == 404 + document = json.loads(result.stdout) + assert document["detail"] == "Not found." + assert document["meta"]["code"] == 404 mock_regenerate.assert_not_called() @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") @@ -951,19 +1027,23 @@ def test_get_not_found(self, mock_list, runner): # AliasGroup.main runs click with standalone_mode=False, so the status # comes back as the command's return value rather than an exit code. assert result.return_value == 404 - assert "Could not get GPG key for my-repo: not found." in result.output + assert "Could not get GPG key for my-org/my-repo: not found." in result.output assert "status: 404" not in result.output @pytest.mark.parametrize( ("status", "expected"), [ - (400, "Could not set GPG key for my-repo: the provided key is not valid."), + ( + 400, + "Could not set GPG key for my-org/my-repo: the provided key is not " + "valid.", + ), ( 402, - "Could not set GPG key for my-repo: custom GPG keys require a paid " - "plan.", + "Could not set GPG key for my-org/my-repo: custom GPG keys require a " + "paid plan.", ), - (404, "Could not set GPG key for my-repo: not found."), + (404, "Could not set GPG key for my-org/my-repo: not found."), ], ) @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") @@ -999,10 +1079,46 @@ def test_regenerate_failure(self, mock_regenerate, runner): assert result.return_value == 402 assert ( - "Could not regenerate GPG key for my-repo: custom GPG keys require a " - "paid plan." in result.output + "Could not regenerate GPG key for my-org/my-repo: custom GPG keys " + "require a paid plan." in result.output ) + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + def test_regenerate_400_keeps_the_standard_rendering(self, mock_regenerate, runner): + """No key is sent by 'regenerate', so a 400 can't mean "bad key".""" + mock_regenerate.side_effect = ApiException( + status=400, detail="Some other validation problem." + ) + + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "-y"), + catch_exceptions=False, + ) + + assert result.return_value == 400 + assert "the provided key is not valid" not in result.output + assert "Detail: Some other validation problem." in result.output + + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") + def test_regenerate_dry_run_400_keeps_the_standard_rendering( + self, mock_list, runner + ): + """The dry-run pre-flight is a GET; a 400 there can't mean "bad key" either.""" + mock_list.side_effect = ApiException( + status=400, detail="Some other validation problem." + ) + + result = runner.invoke( + main, + gpg_command_args("regenerate", "my-org/my-repo", "--dry-run"), + catch_exceptions=False, + ) + + assert result.return_value == 400 + assert "the provided key is not valid" not in result.output + assert "Detail: Some other validation problem." in result.output + @patch("cloudsmith_cli.cli.commands.repos.api.list_repo_gpg_key") def test_unmapped_status_keeps_the_standard_rendering(self, mock_list, runner): mock_list.side_effect = ApiException(status=500, detail="Boom.")