diff --git a/CHANGELOG.md b/CHANGELOG.md index 1125a47d..4b87e660 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added +- Added `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. - Added `cloudsmith repos privileges` for managing explicit repository access from the terminal. `list` shows the teams, users and service accounts that were granted access explicitly; `set` grants access to any number of them and leaves everyone else untouched, asking first if it would lower access someone already has; `revoke` takes access away from the ones named, skipping any that had none; and `replace` makes a JSON file (or stdin) the complete truth for the repository. `revoke` and `replace` ask for confirmation first unless `-y` is passed. ### Changed diff --git a/cloudsmith_cli/cli/commands/repos.py b/cloudsmith_cli/cli/commands/repos.py index 11f43385..4944a74c 100644 --- a/cloudsmith_cli/cli/commands/repos.py +++ b/cloudsmith_cli/cli/commands/repos.py @@ -346,6 +346,491 @@ 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() + + +#: 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_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" + +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." +) + + +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, owner, repo, reasons): + """Build single-line error messages for a GPG command, keyed by status.""" + return { + status: f"Could not {action} GPG key for {owner}/{repo}: {reason}." + for status, reason in reasons.items() + } + + +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. + """ + 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, 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, + { + "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() + + +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 + """ + 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, + error_summaries=gpg_error_summaries( + "get", owner, repo, GPG_READ_ERROR_REASONS + ), + ), + 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=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 " + "file path.", +) +@click.option( + "--passphrase-file", + "passphrase_file", + type=SecretFile("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). 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.option( + "-n", + "--dry-run", + "dry_run", + default=False, + is_flag=True, + 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): + """ + 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. + + Use -n/--dry-run to check the key and passphrase inputs, and the key + currently in place, without changing anything. + + 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) + + 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", + ) + + 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 '-'.", + param_hint="--passphrase-file", + ) + + 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" + ) + + if passphrase_file is not 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] + 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)", + hide_input=True, + default="", + 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: + gpg_dry_run( + ctx, + opts, + "set", + GPG_NO_KEY_ERROR_REASONS, + owner, + repo, + note=passphrase_note, + 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 ... ", + 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, + error_summaries=gpg_error_summaries( + "set", owner, repo, GPG_UPLOAD_ERROR_REASONS + ), + ), + 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.option( + "-n", + "--dry-run", + "dry_run", + default=False, + is_flag=True, + 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): + """ + 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. + + 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 + """ + owner, repo = owner_repo + use_stderr = utils.should_use_stderr(opts) + + if dry_run: + gpg_dry_run( + ctx, + opts, + "regenerate", + GPG_NO_KEY_ERROR_REASONS, + owner, + repo, + 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, + error_summaries=gpg_error_summaries( + "regenerate", owner, repo, GPG_NO_KEY_ERROR_REASONS + ), + ), + 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/exceptions.py b/cloudsmith_cli/cli/exceptions.py index 703e69c6..de7f01f4 100644 --- a/cloudsmith_cli/cli/exceptions.py +++ b/cloudsmith_cli/cli/exceptions.py @@ -17,15 +17,22 @@ def handle_api_exceptions( nl=False, exit_on_error=True, reraise_on_error=False, + error_summaries=None, summarise_error=None, ): """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. + ``summarise_error`` is an optional callable taking ``(exc, detail, fields)`` and returning a single sentence to show instead of the default - context/detail/field block, or ``None`` to keep the default. Commands use - it where the API's field-indexed errors read poorly next to the rest of - their output; returning ``None`` for statuses they don't recognise keeps + context/detail/field block, or ``None`` to keep the default. Unlike + ``error_summaries``, this replaces the JSON ``detail`` too - use it + where the API's field-indexed errors read poorly next to the rest of + their output; returning ``None`` for statuses it doesn't recognise keeps the status code visible where it still matters. """ # flake8: ignore=C901 @@ -83,47 +90,14 @@ def handle_api_exceptions( else: click.secho("ERROR", fg="red", err=use_stderr) + # A command-specific one-liner - from either mechanism - replaces + # the generic context/detail/fields/hint block entirely. + summary = summary or (error_summaries or {}).get(exc.status) if summary: click.secho(summary, fg="red", err=use_stderr) else: - click.secho( - f"{context_msg} (status: {exc.status} - {exc.status_description})", - fg="red", - err=use_stderr, - ) - - if not summary and (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, + print_error_details( + context_msg, exc, detail, fields, hint, use_stderr=use_stderr ) if opts.verbose and not opts.debug and exc.headers: @@ -139,6 +113,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 4d159167..74f32fb9 100644 --- a/cloudsmith_cli/cli/tests/commands/test_repos.py +++ b/cloudsmith_cli/cli/tests/commands/test_repos.py @@ -1,11 +1,43 @@ import json +from unittest.mock import patch 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 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 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.""" @@ -200,3 +232,918 @@ 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") + @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( + main, + gpg_command_args(command_name, "my-org/my-repo"), + 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( + main, + gpg_command_args("get", "my-org/my-repo", "-F", "json"), + 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( + 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") + 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( + 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 + 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_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.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, _is_tty, 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) + ), + 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") + @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.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, _is_tty, 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( + main, + gpg_command_args( + "upload", "my-org/my-repo", "--private-key-file", str(key_file) + ), + 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_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.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + 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" + 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 "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 + 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 "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) + @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") + 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.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.create_repo_gpg_key") + 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) + + result = runner.invoke( + main, + gpg_command_args( + "upload", + "my-org/my-repo", + "--private-key-file", + str(key_file), + "--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": "set", + "namespace": "my-org", + "repository": "my-repo", + "current_fingerprint": _GPG_KEY["fingerprint"], + } + 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( + main, + gpg_command_args("upload", "my-org/my-repo", "--private-key", "sekrit"), + catch_exceptions=False, + ) + + 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) + @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_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"), + 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.list_repo_gpg_key") + @patch("cloudsmith_cli.cli.commands.repos.api.regenerate_repo_gpg_key") + 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"), + catch_exceptions=False, + ) + + 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") + 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_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 "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") + @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"), + 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", + "current_fingerprint": _GPG_KEY["fingerprint"], + } + 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( + main, + gpg_command_args("regenerate", "my-org/my-repo", "-y"), + 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 + + @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 ") + + +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-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-org/my-repo: the provided key is not " + "valid.", + ), + ( + 402, + "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-org/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-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.") + + 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 diff --git a/cloudsmith_cli/cli/tests/conftest.py b/cloudsmith_cli/cli/tests/conftest.py index 6ba3fe94..8e139130 100644 --- a/cloudsmith_cli/cli/tests/conftest.py +++ b/cloudsmith_cli/cli/tests/conftest.py @@ -6,6 +6,7 @@ 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 @@ -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.""" diff --git a/cloudsmith_cli/core/api/repos.py b/cloudsmith_cli/core/api/repos.py index 35b04444..b5f4410f 100644 --- a/cloudsmith_cli/core/api/repos.py +++ b/cloudsmith_cli/core/api/repos.py @@ -81,6 +81,45 @@ def delete_repo(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() + + def list_repo_privileges(owner, repo): """Get the explicit team/user/service privileges on a repository. 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