diff --git a/CHANGELOG.md b/CHANGELOG.md index a4b10ebc..29ff7f23 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -10,6 +10,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. ### Added - Added `-w` and `--workspace` as the shared Workspace option used by authentication, OIDC, and custom-domain discovery. Set `CLOUDSMITH_WORKSPACE` or `workspace` in `config.ini` to configure it once for every command. +- Added a Terraform credentials helper for Cloudsmith registries. `cloudsmith credential-helper install terraform` writes a `terraform-credentials-cloudsmith` launcher into Terraform's plugin directory (`~/.terraform.d/plugins` by default; override with `--bin-dir`) and adds a `credentials_helper "cloudsmith"` block to `~/.terraformrc`, so `terraform init` authenticates against a Cloudsmith Terraform registry with no token on disk, using your existing CLI credentials (env, config, keyring, or OIDC). The resolved `--org` and `-P/--profile` are baked into the terraformrc `args` list, so no environment variables are needed at `terraform init` time. `get` returns `{"token": "..."}` for a Cloudsmith host (including custom domains) and `{}` for any other host so Terraform falls back to its own credential sources; `store`/`forget` are unsupported. Missing credentials for a Cloudsmith host produce an actionable error rather than a traceback. Manage with `cloudsmith credential-helper uninstall terraform` and `cloudsmith credential-helper list`. ### Changed diff --git a/cloudsmith_cli/cli/commands/credential_helper/__init__.py b/cloudsmith_cli/cli/commands/credential_helper/__init__.py index 66c4121c..5a9a06fd 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/__init__.py +++ b/cloudsmith_cli/cli/commands/credential_helper/__init__.py @@ -14,6 +14,7 @@ from .generic import generic as generic_cmd from .manage import install_cmd, list_cmd, uninstall_cmd from .pnpm import pnpm as pnpm_cmd +from .terraform import terraform as terraform_cmd @click.group() @@ -50,5 +51,6 @@ def credential_helper(): credential_helper.add_command(uninstall_cmd, name="uninstall") credential_helper.add_command(list_cmd, name="list") credential_helper.add_command(cargo_cmd, name="cargo") +credential_helper.add_command(terraform_cmd, name="terraform") main.add_command(credential_helper, name="credential-helper") diff --git a/cloudsmith_cli/cli/commands/credential_helper/manage.py b/cloudsmith_cli/cli/commands/credential_helper/manage.py index f2745e3a..2aa615d3 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/manage.py +++ b/cloudsmith_cli/cli/commands/credential_helper/manage.py @@ -15,6 +15,10 @@ from cloudsmith_cli.credential_helpers.cargo.installer import CargoInstaller from cloudsmith_cli.credential_helpers.generic import PartialInstallError from cloudsmith_cli.credential_helpers.pnpm.installer import PNPMInstaller +from cloudsmith_cli.credential_helpers.terraform.installer import TerraformInstaller +from cloudsmith_cli.credential_helpers.terraform.terraformrc import ( + TerraformrcConflictError, +) from ....credential_helpers.docker.installer import DockerInstaller from ... import utils @@ -33,6 +37,7 @@ "docker": DockerInstaller, "pnpm": PNPMInstaller, "cargo": CargoInstaller, + "terraform": TerraformInstaller, } @@ -65,6 +70,59 @@ def _get_installer(name: str): return cls() +def _terraform_helper_args(ctx, opts, repo: str | None = None) -> tuple[str, ...]: + """Build the terraformrc ``args`` list from the resolved org, repo and profile. + + Baking ``--org``/``-r``/``-P`` into the block means ``terraform init`` works + with no environment variables and no hand-edited config. Only values + actually supplied are written, so an install without ``--org``/``--repo``/ + ``-P`` leaves ``args = []``. + """ + args: list[str] = [] + if opts.org: + args.extend(["--org", opts.org]) + if repo: + args.extend(["-r", repo]) + profile = ctx.meta.get("profile") + if profile: + args.extend(["-P", profile]) + return tuple(args) + + +_REPO_FLAGS = ("-r", "--repo", "--repository") + + +def _terraform_next_steps(helper_args: tuple[str, ...]) -> list[str]: + """Return the post-install guidance for the required repository. + + Terraform never tells the credentials helper which repository is being + requested, so the helper needs the repository supplied out-of-band. If it + was not baked into the generated ``args`` at install time, tell the user how + to provide it: set ``CLOUDSMITH_REPO`` or add ``--repo`` to the terraformrc + ``args``. Returns an empty list when a repository is already configured. + """ + has_repo = any( + a in _REPO_FLAGS or a.split("=", 1)[0] in _REPO_FLAGS for a in helper_args + ) + if has_repo: + return [] + return [ + ( + "Next steps: the Terraform helper requires a repository, which" + " Terraform does not pass to credentials helpers. Provide it in one" + " of these ways:" + ), + ( + " - export CLOUDSMITH_REPO= in the environment that runs" + " `terraform init`, or" + ), + ( + " - add it to the args in ~/.terraformrc, e.g." + ' args = ["--repo", "", ...].' + ), + ] + + # --------------------------------------------------------------------------- # install # --------------------------------------------------------------------------- @@ -99,6 +157,16 @@ def _get_installer(name: str): default=False, help="Bypass the custom-domain cache and fetch fresh data from the API.", ) +@click.option( + "-r", + "--repo", + "--repository", + "repo", + default=None, + help="Terraform only: bake this repository into the terraformrc `args` " + "list (as `-r `) so `terraform init` needs no CLOUDSMITH_REPO env " + "var. Terraform does not pass the repository to a credentials helper.", +) @common_cli_config_options @common_cli_output_options @common_api_auth_options @@ -113,6 +181,7 @@ def install_cmd( dry_run: bool, no_discover: bool, refresh: bool, + repo: str | None, ) -> None: """Install a credential helper launcher and configure the package manager. @@ -140,19 +209,35 @@ def install_cmd( \b # Disable automatic custom-domain discovery $ cloudsmith credential-helper install HELPER --no-discover + + \b + # Terraform: bake the repository into the terraformrc args so + # `terraform init` needs no CLOUDSMITH_REPO env var + $ cloudsmith credential-helper install terraform --org acme --repo my-repo """ installer = _get_installer(helper) + + install_kwargs = { + "bin_dir": bin_dir, + "domains": domains, + "dry_run": dry_run, + "discover": not no_discover, + "refresh": refresh, + "org": opts.org, + "credential": opts.credential, + "api_host": opts.api_host, + } + + # Terraform bakes the resolved org/profile into the terraformrc `args` list + # so `terraform init` needs neither env vars nor a hand-edited config. The + # wrapper forwards these to the CLI ahead of the hostname at call time. + if helper == "terraform": + install_kwargs["helper_args"] = _terraform_helper_args(ctx, opts, repo) + try: - actions = installer.install( - bin_dir=bin_dir, - domains=domains, - dry_run=dry_run, - discover=not no_discover, - refresh=refresh, - org=opts.org, - credential=opts.credential, - api_host=opts.api_host, - ) + actions = installer.install(**install_kwargs) + except TerraformrcConflictError as exc: + raise click.ClickException(str(exc)) except OSError as exc: raise click.ClickException( f"Failed to install {helper!r} credential helper: {exc}" @@ -166,11 +251,17 @@ def install_cmd( use_stderr = utils.should_use_stderr(opts) warnings = [a for a in actions if a.startswith("WARNING")] normal = [a for a in actions if not a.startswith("WARNING")] + + next_steps: list[str] = [] + if helper == "terraform": + next_steps = _terraform_next_steps(install_kwargs.get("helper_args", ())) + data = { "helper": helper, "dry_run": dry_run, "actions": normal, "warnings": warnings, + "next_steps": next_steps, } if utils.maybe_print_as_json(opts, data): sys.exit(ec) @@ -181,6 +272,8 @@ def install_cmd( click.echo(f" {action}" if dry_run else action, err=use_stderr) for warning in warnings: click.secho(f" {warning}" if dry_run else warning, err=True, fg="yellow") + for line in next_steps: + click.secho(line, err=use_stderr, fg="cyan") sys.exit(ec) diff --git a/cloudsmith_cli/cli/commands/credential_helper/terraform.py b/cloudsmith_cli/cli/commands/credential_helper/terraform.py new file mode 100644 index 00000000..4d015f7b --- /dev/null +++ b/cloudsmith_cli/cli/commands/credential_helper/terraform.py @@ -0,0 +1,137 @@ +# Copyright 2026 Cloudsmith Ltd +""" +Terraform credentials helper command. + +Implements the ``get`` verb of Terraform's credentials-helper protocol for +Cloudsmith registries. The installed ``terraform-credentials-cloudsmith`` +launcher forwards Terraform's invocation to this command. + +See: https://developer.hashicorp.com/terraform/internals/credentials-helpers +""" + +import sys + +import click + +from ....credential_helpers.terraform import execute +from ...decorators import ( + common_api_auth_options, + common_cli_config_options, + resolve_credentials, +) + + +@click.command(context_settings={"ignore_unknown_options": True}) +@click.option( + "-r", + "--repo", + "--repository", + "repo", + required=True, + envvar="CLOUDSMITH_REPO", + help="The Cloudsmith repository the registry serves. Terraform does not " + "pass the repository to a credentials helper, so it must be configured " + "here (e.g. in the terraformrc args) to build a repository-scoped token.", +) +@click.argument("params", nargs=-1, type=click.UNPROCESSED) +@common_cli_config_options +@common_api_auth_options +@resolve_credentials +def terraform(opts, repo, params): + """ + Terraform credentials helper for Cloudsmith registries. + + Resolves the token for a Cloudsmith Terraform registry and prints it in + Terraform's expected JSON credentials format: ``{"token": "..."}``. + + Provides credentials for all Cloudsmith Terraform registries: + ``*.cloudsmith.io``, ``*.cloudsmith.com``, and any custom domains + configured for the organisation (requires an organisation - ``--org``, + CLOUDSMITH_ORG or ``org`` in ``config.ini`` - and a valid API key/token). + + Accepts Terraform's calling convention — an optional verb (``get``, + ``store``, ``forget``) followed by the hostname — so the on-PATH launcher + can forward Terraform's arguments verbatim. When no verb is given the + action is ``get``. When no hostname is given it is read from stdin. Only + ``get`` is served; ``store``/``forget`` return an error and a non-zero exit. + + A hostname that is not a Cloudsmith registry yields an empty object + (``{}``) and exit 0 so Terraform falls back to its own credential sources. + + \b + Input (arguments or stdin): + [VERB] HOSTNAME — e.g. "get terraform.cloudsmith.io" or just + "terraform.cloudsmith.io"; HOSTNAME alone may also come from stdin. + + \b + Output (stdout): + JSON: {"token": ""} (Cloudsmith host, token found) + JSON: {} (not a Cloudsmith host) + + \b + Exit codes: + 0: Token returned, or the host is not a Cloudsmith registry + 1: Cloudsmith host with no credentials available, or an error occurred + + The token is scoped to the repository. On a standard + ``*.cloudsmith.io``/``*.cloudsmith.com`` host it is ``{org}/{repo}/{token}`` + and the organisation is required (``--org``, ``CLOUDSMITH_ORG`` or ``org`` + in ``config.ini``); a standard host requested without an organisation is a + non-zero exit. On a custom domain — which is already bound to a single + organisation — the org is omitted and the token is ``{repo}/{token}``. The + repository is always required (``-r/--repo/--repository`` or + ``CLOUDSMITH_REPO``): Terraform does not tell a credentials helper which + repository is being requested. A non-Cloudsmith host still returns ``{}`` + and exit 0. The profile can also be supplied with ``-P/--profile``. The + launcher forwards Terraform's ``args`` verbatim, so a terraformrc block such + as ``credentials_helper "cloudsmith" { args = ["--org=acme", + "--repo=my-repo", "-P", "ci"] }`` reaches this command as those options. + + \b + Examples: + # Direct usage + $ cloudsmith credential-helper terraform --repo my-repo terraform.cloudsmith.io + {"token": "..."} + + # Terraform's calling convention (verb + hostname) + $ cloudsmith credential-helper terraform --repo my-repo get terraform.cloudsmith.io + + # Select an org and profile explicitly (no env vars needed) + $ cloudsmith credential-helper terraform --org=acme -P ci get terraform.cloudsmith.io + + \b + Environment variables: + CLOUDSMITH_API_KEY: API key for authentication (optional) + CLOUDSMITH_ORG: Organisation slug (required to scope the token) + CLOUDSMITH_REPO: Repository slug the registry serves (required) + CLOUDSMITH_PROFILE: Configuration profile to load (optional) + """ + # Terraform passes " "; direct/manual use may pass just the + # hostname (verb defaults to "get") or nothing (hostname read from stdin). + verb = "get" + hostname: str | None = None + if len(params) >= 2: + verb, hostname = params[-2], params[-1] + elif len(params) == 1: + hostname = params[0] + + if not hostname: + try: + hostname = sys.stdin.read().strip() + except (OSError, ValueError): + hostname = "" + + exit_code, stdout, stderr = execute( + verb, + hostname, + credential=opts.credential, + api_host=opts.api_host, + org=opts.org, + repo=repo, + ) + + if stdout is not None: + click.echo(stdout) + if stderr is not None: + click.echo(stderr, err=True) + sys.exit(exit_code) diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py index 839f39ce..9e073bb6 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py @@ -14,7 +14,10 @@ from ....core.api.init import initialise_api from ....core.credentials.models import CredentialResult from ....credential_helpers.backends import BackendKind -from ....credential_helpers.common import is_cloudsmith_domain +from ....credential_helpers.common import ( + is_cloudsmith_domain, + is_standard_cloudsmith_domain, +) from ....credential_helpers.custom_domains import ( CACHE_FORMAT_VERSION, CustomDomain, @@ -720,6 +723,31 @@ def test_is_cloudsmith_domain( assert result is expected +@pytest.mark.parametrize( + "url,expected", + [ + # Standard apex + subdomains, any scheme/path/casing → True + ("cloudsmith.io", True), + ("cloudsmith.com", True), + ("docker.cloudsmith.io", True), + ("https://terraform.cloudsmith.io/acme/repo/", True), + ("TERRAFORM.CLOUDSMITH.COM", True), + # Custom domains and foreign hosts → False (never standard) + ("tf.acme.com", False), + ("docker.acme.com", False), + ("evil.example.com", False), + # Lookalikes that must not match the suffix check + ("notcloudsmith.io", False), + ("cloudsmith.io.evil.com", False), + ("", False), + ], +) +def test_is_standard_cloudsmith_domain(url, expected): + """Only *.cloudsmith.io/.com (and the apexes) are standard; custom domains + and lookalikes are not — no API/auth is consulted.""" + assert is_standard_cloudsmith_domain(url) is expected + + # --------------------------------------------------------------------------- # 10. Docker runtime backend_kind wiring # --------------------------------------------------------------------------- diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper_terraform_integration.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper_terraform_integration.py new file mode 100644 index 00000000..77a596f3 --- /dev/null +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_terraform_integration.py @@ -0,0 +1,175 @@ +# Copyright 2026 Cloudsmith Ltd +"""Live integration test for the Terraform credentials helper. + +This is the end-to-end check the unit tests cannot make: that a real +``terraform init`` authenticates against a Cloudsmith Terraform registry using +*only* the ``terraform-credentials-cloudsmith`` helper, with no token on disk. + +It is intentionally minimal — it proves authentication and nothing else. It +does not assert a module is downloaded (that would couple the test to specific +registry contents); it asserts that Terraform's module installation was *not* +rejected for authentication reasons, which is the one thing the helper is +responsible for. + +Requires: + * ``terraform`` on PATH (skipped otherwise). + * ``terraform-credentials-cloudsmith`` on PATH — i.e. cloudsmith-cli + installed as a console script (skipped otherwise). + * ``PYTEST_CLOUDSMITH_API_KEY`` and ``PYTEST_CLOUDSMITH_ORGANIZATION``. + * ``PYTEST_CLOUDSMITH_TERRAFORM_MODULE`` — a module source in a Cloudsmith + Terraform registry, e.g. + ``terraform.cloudsmith.io///`` (skipped otherwise). + * Optionally ``PYTEST_CLOUDSMITH_TERRAFORM_VERSION`` (defaults to a + wide-open constraint). +""" + +from __future__ import annotations + +import os +import shutil +import subprocess +import textwrap + +import pytest + +# Authentication-failure fingerprints in `terraform init` output. If none of +# these appear, the helper supplied a credential Terraform accepted. +_AUTH_FAILURE_MARKERS = ( + "401", + "403", + "unauthorized", + "authentication", + "forbidden", + "invalid token", + "could not retrieve", # Terraform's wording when a creds helper returns nothing +) + + +def _get_env_var_or_skip(key: str) -> str: + value = os.environ.get(key) + if not value: + pytest.skip(f"{key} not provided") + return value + + +@pytest.fixture() +def terraform_bin() -> str: + """Path to the terraform executable, or skip if it is not installed.""" + path = shutil.which("terraform") + if not path: + pytest.skip("terraform is not installed / not on PATH") + return path + + +@pytest.fixture() +def helper_bin() -> str: + """Path to the terraform-credentials-cloudsmith wrapper, or skip.""" + path = shutil.which("terraform-credentials-cloudsmith") + if not path: + pytest.skip( + "terraform-credentials-cloudsmith not on PATH " + "(install cloudsmith-cli as a console script)" + ) + return path + + +@pytest.mark.integration +def test_terraform_init_authenticates_via_helper( + terraform_bin, helper_bin, tmp_path, monkeypatch +): + """A `terraform init` with no token on disk authenticates via the helper. + + The test isolates HOME so the developer's real ``~/.terraformrc`` and + plugin dir are never touched — guaranteeing there is genuinely no token on + disk and that the *only* credential source is our helper. + """ + api_key = _get_env_var_or_skip("PYTEST_CLOUDSMITH_API_KEY") + organization = _get_env_var_or_skip("PYTEST_CLOUDSMITH_ORGANIZATION") + module_source = _get_env_var_or_skip("PYTEST_CLOUDSMITH_TERRAFORM_MODULE") + module_version = os.environ.get("PYTEST_CLOUDSMITH_TERRAFORM_VERSION", ">= 0.0.0") + + # --- Isolated, token-free Terraform environment --------------------------- + fake_home = tmp_path / "home" + plugin_dir = fake_home / ".terraform.d" / "plugins" + plugin_dir.mkdir(parents=True) + + # Terraform only searches its default plugin locations for credentials + # helpers (it ignores -plugin-dir), so drop the wrapper in there. + helper_link = plugin_dir / "terraform-credentials-cloudsmith" + try: + helper_link.symlink_to(helper_bin) + except OSError: + shutil.copy2(helper_bin, helper_link) + helper_link.chmod(0o755) + + # A terraformrc that configures ONLY the helper — deliberately no + # `credentials` block, so Terraform must consult the helper. + terraformrc = fake_home / ".terraformrc" + terraformrc.write_text( + textwrap.dedent( + """ + credentials_helper "cloudsmith" { + args = [] + } + """ + ).strip() + + "\n", + encoding="utf-8", + ) + + # A one-module config: the smallest thing that forces Terraform to + # authenticate against the Cloudsmith Terraform registry during `init`. + workdir = tmp_path / "tf" + workdir.mkdir() + (workdir / "main.tf").write_text( + textwrap.dedent( + f""" + module "auth_probe" {{ + source = "{module_source}" + version = "{module_version}" + }} + """ + ).strip() + + "\n", + encoding="utf-8", + ) + + # Clean, hermetic environment: our fake HOME, our API key for the helper's + # credential chain, and nothing that could smuggle in a token + # (TF_TOKEN_*, an inherited ~/.terraformrc, etc.). + env = { + "HOME": str(fake_home), + "PATH": f"{os.path.dirname(helper_bin)}{os.pathsep}{os.environ['PATH']}", + "CLOUDSMITH_API_KEY": api_key, + "CLOUDSMITH_ORG": organization, + # Make Terraform's own credential sources impossible so a pass can only + # be attributed to the helper. + "TF_CLI_CONFIG_FILE": str(terraformrc), + "CHECKPOINT_DISABLE": "1", + "TF_IN_AUTOMATION": "1", + "TF_LOG": "trace", # so the helper invocation is visible on failure + } + for key in [k for k in os.environ if k.startswith("TF_TOKEN_")]: + env[key] = "" # neutralise any host-specific bearer-token env vars + + result = subprocess.run( + [terraform_bin, "init", "-no-color", "-input=false"], + cwd=str(workdir), + env=env, + capture_output=True, + text=True, + check=False, + ) + + combined = (result.stdout + "\n" + result.stderr).lower() + + # The assertion is specifically about *authentication*, not about whether + # the module resolved. If init failed, it must not have failed for an + # auth reason. + auth_failures = [m for m in _AUTH_FAILURE_MARKERS if m in combined] + assert not auth_failures, ( + "terraform init failed to authenticate via the credentials helper " + f"(matched {auth_failures!r}).\n\n" + f"exit code: {result.returncode}\n\n" + f"stdout:\n{result.stdout}\n\nstderr:\n{result.stderr}" + ) diff --git a/cloudsmith_cli/cli/tests/test_credential_helper_terraform.py b/cloudsmith_cli/cli/tests/test_credential_helper_terraform.py new file mode 100644 index 00000000..0c574368 --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_credential_helper_terraform.py @@ -0,0 +1,457 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the `cloudsmith credential-helper terraform` runtime and CLI shim. + +Terraform's credentials-helper protocol is a one-shot per request: it runs the +helper as ``terraform-credentials-cloudsmith [args...] `` and, +for ``get``, expects either a JSON credentials object (``{"token": "..."}``) or +an empty object (``{}``) on stdout with exit 0, or an end-user-oriented error on +stderr with a non-zero exit. + +These tests pin the three ``get`` outcomes (token / empty-object / refusal) and +the rejection of ``store``/``forget`` and unknown verbs. +""" + +from __future__ import annotations + +import json +from unittest.mock import patch + +import click.testing +import pytest + +from ...core.credentials.models import CredentialResult +from ...credential_helpers.backends import BackendKind +from ...credential_helpers.terraform.runtime import ( + _MISSING_ORG_MESSAGE, + _REFUSAL_MESSAGE, + execute, + get_token, +) +from ..commands.credential_helper.terraform import terraform + +CLOUDSMITH_HOST = "terraform.cloudsmith.io" +FOREIGN_HOST = "registry.terraform.io" + + +@pytest.fixture() +def runner(): + """Return a CliRunner.""" + return click.testing.CliRunner() + + +@pytest.fixture() +def credential(): + """Return a resolved API-key credential.""" + return CredentialResult(api_key="k_abc", source_name="test") + + +# --------------------------------------------------------------------------- +# 1. get_token — the credential path +# --------------------------------------------------------------------------- + + +def test_get_token_returns_api_key_for_cloudsmith_host(credential): + """A Cloudsmith host with a credential yields the raw API key as the token.""" + assert get_token(CLOUDSMITH_HOST, credential=credential) == "k_abc" + + +def test_get_token_accepts_a_full_https_url(credential): + """The hostname argument may include the scheme and path.""" + url = f"https://{CLOUDSMITH_HOST}/acme/repo/" + assert get_token(url, credential=credential) == "k_abc" + + +def test_get_token_uses_the_terraform_backend_kind_for_custom_domains(credential): + """Custom-domain matching is scoped to Terraform-backed domains.""" + with patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ) as mock_check: + get_token("https://tf.acme.com/", credential=credential, org="acme") + + assert mock_check.call_args.kwargs["backend_kind"] is BackendKind.TERRAFORM + assert mock_check.call_args.kwargs["org"] == "acme" + + +# --------------------------------------------------------------------------- +# 2. get_token — the "no token" paths +# --------------------------------------------------------------------------- + + +def test_get_token_returns_none_without_a_credential(): + """A missing credential means no token, and no exception.""" + assert get_token(CLOUDSMITH_HOST, credential=None) is None + + +def test_get_token_returns_none_for_a_credential_without_an_api_key(): + """An empty api_key is treated as no credential at all.""" + empty = CredentialResult(api_key="", source_name="test") + assert get_token(CLOUDSMITH_HOST, credential=empty) is None + + +def test_get_token_returns_none_for_a_foreign_host(credential): + """A non-Cloudsmith host gets no token — the helper must not leak it.""" + assert get_token(FOREIGN_HOST, credential=credential) is None + + +# --------------------------------------------------------------------------- +# 3. execute('get', ...) — the tuple contract +# --------------------------------------------------------------------------- + + +def test_execute_get_returns_token_object_for_cloudsmith_host(credential): + """Happy path: exit-0, ``{"token": ...}`` on stdout, no stderr. + + The token is scoped to the owner/repository as ``{owner}/{repo}/{token}``. + """ + exit_code, stdout, stderr = execute( + "get", CLOUDSMITH_HOST, credential=credential, org="acme", repo="myrepo" + ) + + assert (exit_code, stderr) == (0, None) + assert json.loads(stdout) == {"token": "acme/myrepo/k_abc"} + + +def test_execute_get_omits_org_from_token_for_a_custom_domain(credential): + """A custom domain is already bound to one organisation, so its token must + omit the org and be scoped as ``{repo}/{token}`` (not ``{org}/{repo}/...``).""" + with patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ): + exit_code, stdout, stderr = execute( + "get", + "https://tf.acme.com/", + credential=credential, + org="acme", + repo="myrepo", + ) + + assert (exit_code, stderr) == (0, None) + assert json.loads(stdout) == {"token": "myrepo/k_abc"} + + +def test_execute_get_custom_domain_does_not_require_an_org_in_the_token(credential): + """The org is only needed to *resolve* a custom domain, never in its token: + once matched, a custom-domain token is ``{repo}/{token}`` with no org.""" + with patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ): + exit_code, stdout, stderr = execute( + "get", + "https://tf.acme.com/", + credential=credential, + org=None, + repo="myrepo", + ) + + assert (exit_code, stderr) == (0, None) + assert json.loads(stdout) == {"token": "myrepo/k_abc"} + + +def test_execute_get_returns_empty_object_for_foreign_host(credential): + """A non-Cloudsmith host is not ours to answer: emit ``{}`` and exit 0 so + Terraform falls back to its own credential sources.""" + exit_code, stdout, stderr = execute("get", FOREIGN_HOST, credential=credential) + + assert (exit_code, stderr) == (0, None) + assert json.loads(stdout) == {} + + +def test_execute_get_refuses_cloudsmith_host_without_credentials(): + """A Cloudsmith host we can't authenticate is a definitive failure: exit 1 + with an actionable error on stderr.""" + with patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ): + exit_code, stdout, stderr = execute( + "get", CLOUDSMITH_HOST, credential=None, org="acme", repo="myrepo" + ) + + assert (exit_code, stdout, stderr) == (1, None, _REFUSAL_MESSAGE) + + +def test_execute_get_refuses_cloudsmith_host_without_an_org(credential): + """A Cloudsmith host needs an org to build the scoped token: exit 1 with an + actionable error on stderr, not a ``None/repo/token`` credential.""" + exit_code, stdout, stderr = execute( + "get", CLOUDSMITH_HOST, credential=credential, org=None, repo="myrepo" + ) + + assert (exit_code, stdout, stderr) == (1, None, _MISSING_ORG_MESSAGE) + + +def test_execute_get_returns_empty_object_for_foreign_host_without_an_org(credential): + """A foreign host still falls back cleanly (``{}``, exit 0) even with no org + — the org is only required on the Cloudsmith-host path.""" + exit_code, stdout, stderr = execute( + "get", FOREIGN_HOST, credential=credential, org=None, repo="myrepo" + ) + + assert (exit_code, stderr) == (0, None) + assert json.loads(stdout) == {} + + +def test_execute_get_refuses_when_no_hostname_provided(credential): + """An empty hostname is an error, not a traceback.""" + exit_code, stdout, stderr = execute("get", "", credential=credential) + + assert exit_code == 1 + assert stdout is None + assert "hostname" in stderr.lower() + + +def test_execute_get_degrades_cleanly_on_domain_lookup_failure(credential): + """A network/SDK error during custom-domain discovery must not raise.""" + with patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + side_effect=RuntimeError("boom"), + ): + exit_code, stdout, stderr = execute( + "get", "https://tf.acme.com/", credential=credential, org="acme" + ) + + assert (exit_code, stdout) == (1, None) + assert stderr == _REFUSAL_MESSAGE + + +# --------------------------------------------------------------------------- +# 4. execute — store / forget / unknown verbs +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("verb", ["store", "forget"]) +def test_execute_rejects_store_and_forget(verb, credential): + """store/forget are unsupported: there is nothing to store or forget.""" + exit_code, stdout, stderr = execute(verb, CLOUDSMITH_HOST, credential=credential) + + assert exit_code == 1 + assert stdout is None + assert verb in stderr + + +def test_execute_rejects_unknown_verb(credential): + """An unknown verb errors out per the forward-compatibility requirement.""" + exit_code, stdout, stderr = execute( + "frobnicate", CLOUDSMITH_HOST, credential=credential + ) + + assert exit_code == 1 + assert stdout is None + assert "Unknown verb" in stderr + + +# --------------------------------------------------------------------------- +# 5. CLI shim +# --------------------------------------------------------------------------- + + +def test_cli_prints_token_object_for_cloudsmith_host(runner): + """The click shim resolves the credential and prints the JSON token object.""" + result = runner.invoke( + terraform, + args=["-k", "k_abc", "--org", "acme", "--repo", "myrepo", CLOUDSMITH_HOST], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == {"token": "acme/myrepo/k_abc"} + + +def test_cli_accepts_terraform_verb_and_hostname(runner): + """Terraform's `get ` convention is accepted (verb + hostname). + + The on-PATH launcher forwards Terraform's arguments verbatim — including + the `get` verb — so the command must accept the verb rather than treating + it as an unexpected extra positional. + """ + result = runner.invoke( + terraform, + args=[ + "-k", + "k_abc", + "--org", + "acme", + "--repo", + "myrepo", + "get", + CLOUDSMITH_HOST, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == {"token": "acme/myrepo/k_abc"} + + +def test_cli_rejects_store_verb(runner): + """`store ` is answered with a non-zero exit and no token.""" + result = runner.invoke( + terraform, + args=["-k", "k_abc", "--repo", "myrepo", "store", CLOUDSMITH_HOST], + catch_exceptions=False, + ) + + assert result.exit_code == 1 + assert "k_abc" not in result.stdout + + +def test_cli_reads_hostname_from_stdin_when_no_argument(runner): + """Omitting the hostname argument falls back to reading it from stdin.""" + result = runner.invoke( + terraform, + args=["-k", "k_abc", "--org", "acme", "--repo", "myrepo"], + input=f"{CLOUDSMITH_HOST}\n", + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == {"token": "acme/myrepo/k_abc"} + + +def test_cli_accepts_org_and_profile_flags(runner): + """`--org` and `-P/--profile` are accepted and the org reaches the runtime. + + These are the options the wrapper forwards from a terraformrc `args` block, + so they must parse on the command without requiring environment variables. + """ + with patch( + "cloudsmith_cli.cli.commands.credential_helper.terraform.execute", + return_value=(0, '{"token": "k_abc"}', None), + ) as mock_execute: + result = runner.invoke( + terraform, + args=[ + "-k", + "k_abc", + "--repo", + "myrepo", + "--org=acme", + "-P", + "ci", + CLOUDSMITH_HOST, + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert mock_execute.call_args.kwargs["org"] == "acme" + assert mock_execute.call_args.kwargs["repo"] == "myrepo" + + +@pytest.mark.parametrize( + "repo_flag", + [ + ["-r", "myrepo"], + ["--repo", "myrepo"], + ["--repository", "myrepo"], + ["--repo=myrepo"], + ], +) +def test_cli_accepts_repo_flag_forms(runner, repo_flag): + """`-r/--repo/--repository` (and the `=` form) all parse and satisfy the + required repository option.""" + result = runner.invoke( + terraform, + args=["-k", "k_abc", "--org", "acme", *repo_flag, "get", CLOUDSMITH_HOST], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == {"token": "acme/myrepo/k_abc"} + + +def test_cli_repo_is_required(runner): + """The repository is required: omitting it is a usage error, not a token. + + Terraform never tells the helper which repository is requested, so the + command must fail fast rather than emit a credential for an unknown + repository. + """ + result = runner.invoke( + terraform, + args=["-k", "k_abc", CLOUDSMITH_HOST], + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "repo" in result.output.lower() + + +def test_cli_org_is_required_for_a_cloudsmith_host(runner): + """A Cloudsmith host without an org is a non-zero exit, not a token. + + The token is scoped as ``{org}/{repo}/{token}``, so an org must be + configured (``--org``/``CLOUDSMITH_ORG``/``org`` in config.ini) before a + credential can be emitted. + """ + with patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ): + result = runner.invoke( + terraform, + args=["-k", "k_abc", "--repo", "myrepo", "get", CLOUDSMITH_HOST], + env={"CLOUDSMITH_ORG": ""}, + catch_exceptions=False, + ) + + assert result.exit_code == 1 + # No token must leak on stdout on the refusal path. + assert result.stdout == "" + assert "organisation" in result.output.lower() + + +def test_cli_repo_from_env_var(runner): + """CLOUDSMITH_REPO satisfies the required repository without the flag.""" + result = runner.invoke( + terraform, + args=["-k", "k_abc", "--org", "acme", "get", CLOUDSMITH_HOST], + env={"CLOUDSMITH_REPO": "envrepo"}, + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == {"token": "acme/envrepo/k_abc"} + + +def test_cli_prints_empty_object_for_foreign_host(runner): + """A non-Cloudsmith host is exit-0 with ``{}`` and no leaked token.""" + result = runner.invoke( + terraform, + args=["-k", "k_abc", "--repo", "myrepo", FOREIGN_HOST], + catch_exceptions=False, + ) + + assert result.exit_code == 0 + assert json.loads(result.stdout) == {} + assert "k_abc" not in result.stdout + + +def test_cli_exits_non_zero_with_a_hint_when_no_credential_resolves(runner): + """A Cloudsmith host with no resolvable credential is exit-1 with a hint.""" + # Patch the provider chain rather than relying on env vars: a developer's + # local ~/.cloudsmith/config.ini (or an active profile) can otherwise + # resolve a credential and turn this into a false negative. + with ( + patch( + "cloudsmith_cli.cli.decorators.CredentialProviderChain.resolve", + return_value=None, + ), + patch( + "cloudsmith_cli.credential_helpers.terraform.runtime.is_cloudsmith_domain", + return_value=True, + ), + ): + result = runner.invoke( + terraform, + args=["--repo", "myrepo", CLOUDSMITH_HOST], + env={"CLOUDSMITH_API_KEY": ""}, + catch_exceptions=False, + ) + + assert result.exit_code == 1 + # No token must leak on stdout on the refusal path. + assert result.stdout == "" diff --git a/cloudsmith_cli/cli/tests/test_credential_helper_terraform_installer.py b/cloudsmith_cli/cli/tests/test_credential_helper_terraform_installer.py new file mode 100644 index 00000000..0bcc18d6 --- /dev/null +++ b/cloudsmith_cli/cli/tests/test_credential_helper_terraform_installer.py @@ -0,0 +1,381 @@ +# Copyright 2026 Cloudsmith Ltd +"""Tests for the Terraform credentials-helper installer and terraformrc block. + +Covers the pure ``terraformrc`` block helpers (add/update/remove/conflict), the +``TerraformInstaller`` (launcher into the plugin dir + terraformrc block), and +the ``credential-helper install/uninstall terraform`` CLI wiring — including +that the resolved ``--org``/``-P`` land in the terraformrc ``args`` list. +""" + +from __future__ import annotations + +import json +from pathlib import Path + +import click.testing +import pytest + +from ...credential_helpers.terraform import terraformrc +from ...credential_helpers.terraform.installer import TerraformInstaller + +LAUNCHER = "terraform-credentials-cloudsmith" + + +def _launcher(home: Path) -> Path: + """Return the default launcher path under a fake *home*.""" + return home / ".terraform.d" / "plugins" / LAUNCHER + + +@pytest.fixture() +def runner(): + """Return a CliRunner.""" + return click.testing.CliRunner() + + +# --------------------------------------------------------------------------- +# 1. terraformrc block helpers (pure text) +# --------------------------------------------------------------------------- + + +def test_render_block_formats_args(): + """render_block emits a valid HCL block with the args quoted.""" + block = terraformrc.render_block(["--org", "acme", "-P", "ci"]) + assert block == ( + 'credentials_helper "cloudsmith" {\n args = ["--org", "acme", "-P", "ci"]\n}' + ) + + +def test_render_block_empty_args(): + """No args renders an empty list, not a missing key.""" + assert "args = []" in terraformrc.render_block([]) + + +def test_add_block_to_empty_file(): + """Adding to an empty file yields just the block plus a trailing newline.""" + new_text, changed = terraformrc.add_or_update_block("", ["--org", "acme"]) + assert changed is True + assert new_text == terraformrc.render_block(["--org", "acme"]) + "\n" + + +def test_add_block_preserves_foreign_content(): + """Existing settings are kept and the block is appended after a blank line.""" + existing = 'plugin_cache_dir = "/tmp/x"\ndisable_checkpoint = true\n' + new_text, changed = terraformrc.add_or_update_block(existing, []) + assert changed is True + assert new_text.startswith(existing) + assert 'credentials_helper "cloudsmith"' in new_text + + +def test_update_replaces_existing_cloudsmith_block(): + """Re-adding with different args replaces the block in place.""" + first, _ = terraformrc.add_or_update_block("", ["--org", "acme"]) + second, changed = terraformrc.add_or_update_block(first, ["--org", "other"]) + assert changed is True + assert second.count('credentials_helper "cloudsmith"') == 1 + assert '"other"' in second + assert '"acme"' not in second + + +def test_add_is_idempotent(): + """Adding the identical block twice reports no change the second time.""" + first, _ = terraformrc.add_or_update_block("", ["--org", "acme"]) + second, changed = terraformrc.add_or_update_block(first, ["--org", "acme"]) + assert changed is False + assert second == first + + +def test_add_raises_on_foreign_credentials_helper(): + """A credentials_helper for a different helper is a hard conflict.""" + existing = 'credentials_helper "vault" {\n args = []\n}\n' + with pytest.raises(terraformrc.TerraformrcConflictError) as exc: + terraformrc.add_or_update_block(existing, []) + assert exc.value.existing_name == "vault" + + +def test_remove_block_strips_only_cloudsmith(): + """remove_block drops the Cloudsmith block and collapses stray blank lines.""" + existing = ( + 'plugin_cache_dir = "/tmp/x"\n\n' + 'credentials_helper "cloudsmith" {\n args = []\n}\n' + ) + new_text, changed = terraformrc.remove_block(existing) + assert changed is True + assert "credentials_helper" not in new_text + assert new_text == 'plugin_cache_dir = "/tmp/x"\n' + + +def test_remove_block_leaves_foreign_helper_untouched(): + """A different helper's block is not removed.""" + existing = 'credentials_helper "vault" {\n args = []\n}\n' + new_text, changed = terraformrc.remove_block(existing) + assert changed is False + assert new_text == existing + + +def test_remove_block_no_block_is_noop(): + """Removing from a file without our block reports no change.""" + new_text, changed = terraformrc.remove_block("disable_checkpoint = true\n") + assert changed is False + + +# --------------------------------------------------------------------------- +# 2. TerraformInstaller.install / uninstall / status +# --------------------------------------------------------------------------- + + +def test_installer_install_writes_launcher_and_block(tmp_path, monkeypatch): + """install writes the launcher into the plugin dir and the terraformrc block.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + installer = TerraformInstaller() + actions = installer.install(helper_args=("--org", "acme", "-P", "ci")) + + launcher = _launcher(tmp_path) + assert launcher.exists() + body = launcher.read_text(encoding="utf-8") + assert "exec cloudsmith credential-helper terraform" in body + + rc = (tmp_path / ".terraformrc").read_text(encoding="utf-8") + assert 'credentials_helper "cloudsmith"' in rc + assert 'args = ["--org", "acme", "-P", "ci"]' in rc + assert any("wrote launcher" in a for a in actions) + + +def test_installer_respects_bin_dir_override(tmp_path, monkeypatch): + """--bin-dir overrides the default plugin directory for the launcher.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + custom = tmp_path / "custom_plugins" + + installer = TerraformInstaller() + installer.install(bin_dir=str(custom)) + + assert (custom / "terraform-credentials-cloudsmith").exists() + # The default plugin dir must NOT have been used. + assert not (tmp_path / ".terraform.d" / "plugins").exists() + + +def test_installer_dry_run_writes_nothing(tmp_path, monkeypatch): + """dry_run reports planned actions without touching the filesystem.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + installer = TerraformInstaller() + actions = installer.install(helper_args=("--org", "acme"), dry_run=True) + + assert not (tmp_path / ".terraformrc").exists() + assert not (tmp_path / ".terraform.d" / "plugins").exists() + assert any("would write launcher" in a for a in actions) + assert any("would add" in a for a in actions) + + +def test_installer_idempotent(tmp_path, monkeypatch): + """A second install reports the terraformrc is already up to date.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + installer = TerraformInstaller() + installer.install(helper_args=("--org", "acme")) + actions = installer.install(helper_args=("--org", "acme")) + + assert any("already up to date" in a for a in actions) + + +def test_installer_uninstall_removes_launcher_and_block(tmp_path, monkeypatch): + """uninstall removes the launcher and the terraformrc block.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + installer = TerraformInstaller() + installer.install(helper_args=("--org", "acme")) + launcher = _launcher(tmp_path) + assert launcher.exists() + + installer.uninstall() + + assert not launcher.exists() + rc = (tmp_path / ".terraformrc").read_text(encoding="utf-8") + assert "credentials_helper" not in rc + + +def test_installer_status_type_contract(tmp_path, monkeypatch): + """status()['launcher'] is str when installed and None when not — never Path.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + installer = TerraformInstaller() + + before = installer.status() + assert before["launcher"] is None + assert before["hosts"] == [] + + installer.install(helper_args=("--org", "acme")) + after = installer.status() + assert isinstance(after["launcher"], str) + assert after["launcher"].endswith("terraform-credentials-cloudsmith") + assert after["hosts"] # non-empty marker + + +# --------------------------------------------------------------------------- +# 3. CLI wiring — install/uninstall terraform +# --------------------------------------------------------------------------- + + +def test_cli_install_bakes_org_and_profile_into_args(runner, tmp_path, monkeypatch): + """`install terraform --org --P` writes those into the terraformrc args list.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + from ...cli.commands.credential_helper.manage import install_cmd + + result = runner.invoke( + install_cmd, + [ + "terraform", + "--org=acme", + "-P", + "ci", + "--no-discover", + "-k", + "k_flag", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + rc = (tmp_path / ".terraformrc").read_text(encoding="utf-8") + assert 'args = ["--org", "acme", "-P", "ci"]' in rc + + +@pytest.mark.parametrize( + "repo_flag", + [ + ["-r", "my-repo"], + ["--repo", "my-repo"], + ["--repository", "my-repo"], + ["--repo=my-repo"], + ], +) +def test_cli_install_bakes_repo_into_args(runner, tmp_path, monkeypatch, repo_flag): + """`install terraform --repo` writes `-r ` into the terraformrc args.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + from ...cli.commands.credential_helper.manage import install_cmd + + result = runner.invoke( + install_cmd, + ["terraform", "--org=acme", *repo_flag, "--no-discover", "-k", "k_flag"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + rc = (tmp_path / ".terraformrc").read_text(encoding="utf-8") + assert 'args = ["--org", "acme", "-r", "my-repo"]' in rc + + +def test_cli_install_with_repo_suppresses_next_steps(runner, tmp_path, monkeypatch): + """A baked-in repository means the repository guidance is not printed.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + from ...cli.commands.credential_helper.manage import install_cmd + + result = runner.invoke( + install_cmd, + [ + "terraform", + "--org=acme", + "--repo", + "my-repo", + "--no-discover", + "-k", + "k_flag", + ], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + assert "Next steps" not in result.output + assert "CLOUDSMITH_REPO" not in result.output + + +def test_cli_install_conflict_is_clean_error(runner, tmp_path, monkeypatch): + """A pre-existing foreign credentials_helper yields a ClickException, not a traceback.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + (tmp_path / ".terraformrc").write_text( + 'credentials_helper "vault" {\n args = []\n}\n', encoding="utf-8" + ) + + from ...cli.commands.credential_helper.manage import install_cmd + + result = runner.invoke( + install_cmd, + ["terraform", "--no-discover", "-k", "k_flag"], + catch_exceptions=False, + ) + + assert result.exit_code != 0 + assert "only one credentials_helper" in result.output + # The launcher must not have been written when the terraformrc conflicts. + launcher = _launcher(tmp_path) + assert not launcher.exists() + + +def test_cli_install_prints_repo_next_steps(runner, tmp_path, monkeypatch): + """install terraform prints guidance about the required repository.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + from ...cli.commands.credential_helper.manage import install_cmd + + result = runner.invoke( + install_cmd, + ["terraform", "--org=acme", "-P", "ci", "--no-discover", "-k", "k_flag"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + out = result.output + assert "Next steps" in out + assert "CLOUDSMITH_REPO" in out + assert "--repo" in out + + +def test_cli_install_repo_next_steps_in_json(runner, tmp_path, monkeypatch): + """The repository guidance is surfaced as a next_steps field in JSON mode.""" + monkeypatch.setattr(Path, "home", staticmethod(lambda: tmp_path)) + monkeypatch.delenv("TF_CLI_CONFIG_FILE", raising=False) + + from ...cli.commands.credential_helper.manage import install_cmd + + result = runner.invoke( + install_cmd, + ["terraform", "--no-discover", "-k", "k_flag", "-F", "json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + next_steps = payload["data"]["next_steps"] + assert next_steps + assert any("CLOUDSMITH_REPO" in line for line in next_steps) + + +def test_cli_install_docker_has_no_next_steps(runner, tmp_path, monkeypatch): + """Non-terraform helpers do not emit the terraform repository guidance.""" + monkeypatch.setenv("DOCKER_CONFIG", str(tmp_path / ".docker")) + + from ...cli.commands.credential_helper.manage import install_cmd + + result = runner.invoke( + install_cmd, + ["docker", "--no-discover", "-k", "k_flag", "-F", "json"], + catch_exceptions=False, + ) + + assert result.exit_code == 0, result.output + payload = json.loads(result.output) + assert payload["data"]["next_steps"] == [] diff --git a/cloudsmith_cli/cli/tests/test_startup_imports.py b/cloudsmith_cli/cli/tests/test_startup_imports.py index 122b688f..e5a89fb2 100644 --- a/cloudsmith_cli/cli/tests/test_startup_imports.py +++ b/cloudsmith_cli/cli/tests/test_startup_imports.py @@ -62,3 +62,8 @@ def test_cli_import_does_not_load_command_modules(): def test_docker_helper_import_does_not_load_heavy_modules(): modules = modules_loaded_by_import("cloudsmith_cli.credential_helpers.docker") assert heavy_modules_in(modules) == [] + + +def test_terraform_helper_import_does_not_load_heavy_modules(): + modules = modules_loaded_by_import("cloudsmith_cli.credential_helpers.terraform") + assert heavy_modules_in(modules) == [] diff --git a/cloudsmith_cli/credential_helpers/common.py b/cloudsmith_cli/credential_helpers/common.py index e036799a..d79bd0ab 100644 --- a/cloudsmith_cli/credential_helpers/common.py +++ b/cloudsmith_cli/credential_helpers/common.py @@ -46,6 +46,28 @@ def extract_hostname(url): return hostname +def is_standard_cloudsmith_domain(url): + """ + Check if a URL points to a standard Cloudsmith domain. + + Standard domains are the ``*.cloudsmith.io``/``*.cloudsmith.com`` registries + (and the bare apex domains). Custom domains configured for an organisation + are *not* standard domains. + + Args: + url: URL or hostname to check + + Returns: + bool: True if this is a standard Cloudsmith domain + """ + hostname = extract_hostname(url) + if not hostname: + return False + return hostname in ("cloudsmith.io", "cloudsmith.com") or hostname.endswith( + (".cloudsmith.io", ".cloudsmith.com") + ) + + def is_cloudsmith_domain( url, credential=None, api_host=None, backend_kind=None, org=None ): @@ -74,9 +96,7 @@ def is_cloudsmith_domain( return False # Standard Cloudsmith domains — no auth needed, always match regardless of backend_kind - if hostname in ("cloudsmith.io", "cloudsmith.com") or hostname.endswith( - (".cloudsmith.io", ".cloudsmith.com") - ): + if is_standard_cloudsmith_domain(hostname): return True # Custom domains require org + auth diff --git a/cloudsmith_cli/credential_helpers/terraform/__init__.py b/cloudsmith_cli/credential_helpers/terraform/__init__.py new file mode 100644 index 00000000..3658d168 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/terraform/__init__.py @@ -0,0 +1,4 @@ +# Copyright 2026 Cloudsmith Ltd +from .runtime import execute, get_token + +__all__ = ["execute", "get_token"] diff --git a/cloudsmith_cli/credential_helpers/terraform/installer.py b/cloudsmith_cli/credential_helpers/terraform/installer.py new file mode 100644 index 00000000..a1d27d42 --- /dev/null +++ b/cloudsmith_cli/credential_helpers/terraform/installer.py @@ -0,0 +1,289 @@ +# Copyright 2026 Cloudsmith Ltd +"""Installer for the Terraform credentials helper. + +Manages writing/removing the ``terraform-credentials-cloudsmith`` launcher and +the ``credentials_helper "cloudsmith"`` block in ``~/.terraformrc`` so Terraform +authenticates to Cloudsmith Terraform registries through the Cloudsmith +credential chain. + +Unlike Docker/Cargo/pnpm, Terraform does **not** search ``PATH`` for credentials +helpers — it only looks in its default plugin locations. The launcher is +therefore written into ``~/.terraform.d/plugins`` by default (override with +``--bin-dir``) rather than a PATH directory. +""" + +from __future__ import annotations + +import logging +import os +import sys +from pathlib import Path + +from ..launchers import remove_launcher, write_launcher +from . import terraformrc + +logger = logging.getLogger(__name__) + + +def _terraformrc_path() -> Path: + """Return the path to Terraform's CLI configuration file. + + Respects ``TF_CLI_CONFIG_FILE`` (Terraform's own override); otherwise + returns the platform default. On Windows that is ``%APPDATA%/terraform.rc``; + everywhere else it is ``~/.terraformrc``. + """ + override = os.environ.get("TF_CLI_CONFIG_FILE") + if override: + return Path(override) + if os.name == "nt": + appdata = os.environ.get("APPDATA") + base = Path(appdata) if appdata else Path.home() + return base / "terraform.rc" + return Path.home() / ".terraformrc" + + +def _default_plugin_dir() -> Path: + """Return Terraform's conventional user plugin directory. + + ``~/.terraform.d/plugins`` on all platforms — the best-known of Terraform's + default credentials-helper search locations and writable without elevation. + """ + return Path.home() / ".terraform.d" / "plugins" + + +class TerraformInstaller: + """Manages installation of the Terraform credentials helper for Cloudsmith. + + Writes a ``terraform-credentials-cloudsmith`` launcher into Terraform's + plugin directory and adds a ``credentials_helper "cloudsmith"`` block to + ``~/.terraformrc``. + + Usage:: + + installer = TerraformInstaller() + actions = installer.install(helper_args=["--org", "acme"]) + for action in actions: + print(action) + """ + + LAUNCHER_NAME = "terraform-credentials-cloudsmith" + TARGET_CMD = "cloudsmith credential-helper terraform" + + name = "terraform" + summary = "Terraform credentials helper for Cloudsmith registries" + + @classmethod + def _resolve_target_cmd(cls) -> str: + """Return the command the launcher forwards to. + + A pip/source install resolves the bare ``cloudsmith`` command via + ``PATH``. A frozen standalone binary (PyInstaller) is not guaranteed to + be on ``PATH`` under that name, so point the launcher at the absolute + executable instead. The path is quoted so a directory containing spaces + still execs correctly. + """ + if getattr(sys, "frozen", False): + return f'"{sys.executable}" credential-helper terraform' + return cls.TARGET_CMD + + def _resolve_bin_dir(self, bin_dir: str | None) -> Path: + """Return the directory to install the launcher into. + + Defaults to Terraform's user plugin directory (Terraform ignores + ``PATH`` for credentials helpers); an explicit *bin_dir* override is + resolved to an absolute path. + """ + if bin_dir is not None: + return Path(bin_dir).resolve() + return _default_plugin_dir() + + def _launcher_path(self, target_dir: Path) -> Path: + """Return the launcher path within *target_dir* for this platform.""" + if os.name == "nt": + return target_dir / f"{self.LAUNCHER_NAME}.cmd" + return target_dir / self.LAUNCHER_NAME + + # ------------------------------------------------------------------ + # install / uninstall / status + # ------------------------------------------------------------------ + + def install( + self, + *, + bin_dir: str | None = None, + helper_args: tuple[str, ...] = (), + dry_run: bool = False, + # Accepted for a uniform installer interface; Terraform's helper is + # host-agnostic (Terraform passes the hostname at call time). + **kwargs, + ) -> list[str]: + """Install the Terraform credentials helper. + + Writes the launcher into Terraform's plugin directory and adds the + ``credentials_helper "cloudsmith"`` block to ``~/.terraformrc``. + + Parameters + ---------- + bin_dir: + Override for the plugin directory to install the launcher into. + Defaults to ``~/.terraform.d/plugins``. + helper_args: + Values for the block's ``args`` list, forwarded to the CLI on every + invocation — e.g. ``("--org", "acme", "-P", "ci")`` to pin the + organisation and profile without environment variables. + dry_run: + When ``True``, compute and return planned actions without writing. + + Returns + ------- + list[str] + Human-readable descriptions of actions taken (or planned). + """ + target_dir = self._resolve_bin_dir(bin_dir) + launcher_path = self._launcher_path(target_dir) + rc_path = _terraformrc_path() + + actions: list[str] = [] + + existing = "" + rc_exists = rc_path.exists() + if rc_exists: + existing = rc_path.read_text(encoding="utf-8") + + # Compute the terraformrc change up front so a conflict aborts before we + # write the launcher — leaving no orphan behind. + new_rc, rc_changed = terraformrc.add_or_update_block(existing, helper_args) + + if dry_run: + actions.append(f"would write launcher {launcher_path}") + if rc_changed: + verb = "add" if terraformrc.find_block(existing) is None else "update" + actions.append( + f'would {verb} credentials_helper "cloudsmith" in {rc_path}' + ) + else: + actions.append( + f'credentials_helper "cloudsmith" already up to date' + f" in {rc_path} (no change)" + ) + return actions + + # Real install: launcher first, then terraformrc. + written = write_launcher( + target_dir, self.LAUNCHER_NAME, self._resolve_target_cmd() + ) + actions.append(f"wrote launcher {written}") + + if rc_changed: + rc_path.parent.mkdir(parents=True, exist_ok=True) + rc_path.write_text(new_rc, encoding="utf-8") + verb = "added" if not rc_exists else "updated" + actions.append(f'{verb} credentials_helper "cloudsmith" in {rc_path}') + else: + actions.append(f"{rc_path} already up to date") + + return actions + + def uninstall( + self, *, bin_dir: str | None = None, dry_run: bool = False + ) -> list[str]: + """Uninstall the Terraform credentials helper. + + Removes the launcher and strips the ``credentials_helper "cloudsmith"`` + block from ``~/.terraformrc`` (a block for a different helper is left + untouched). + + Parameters + ---------- + bin_dir: + Override for the plugin directory the launcher was installed into. + Pass the same value given to :meth:`install`. + dry_run: + When ``True``, return planned actions without writing. + + Returns + ------- + list[str] + Human-readable descriptions of actions taken (or planned). + """ + target_dir = self._resolve_bin_dir(bin_dir) + launcher_path = self._launcher_path(target_dir) + rc_path = _terraformrc_path() + + actions: list[str] = [] + + existing = "" + rc_exists = rc_path.exists() + if rc_exists: + existing = rc_path.read_text(encoding="utf-8") + new_rc, rc_changed = terraformrc.remove_block(existing) + + if dry_run: + if launcher_path.exists(): + actions.append(f"would remove launcher {launcher_path}") + else: + actions.append( + f"launcher not found at {launcher_path} (nothing to remove)" + ) + if rc_changed: + actions.append( + f'would remove credentials_helper "cloudsmith" from {rc_path}' + ) + else: + actions.append( + f'no credentials_helper "cloudsmith" block to remove from {rc_path}' + ) + return actions + + removed = remove_launcher(target_dir, self.LAUNCHER_NAME) + if removed: + actions.append(f"removed launcher {launcher_path}") + else: + actions.append(f"launcher not found at {launcher_path} (nothing to remove)") + + if rc_changed: + rc_path.write_text(new_rc, encoding="utf-8") + actions.append(f'removed credentials_helper "cloudsmith" from {rc_path}') + else: + actions.append( + f'no credentials_helper "cloudsmith" block to remove from {rc_path}' + ) + + return actions + + def status(self) -> dict: + """Return current installation status. + + Returns + ------- + dict + A dict with keys: + + ``"launcher"`` + The path of the launcher if it exists, else ``None``. + ``"hosts"`` + A single marker when the ``credentials_helper "cloudsmith"`` + block is present in ``~/.terraformrc`` (the helper is + host-agnostic — Terraform passes the hostname at call time), or + an empty list. + """ + target_dir = self._resolve_bin_dir(None) + launcher_path: Path | None = self._launcher_path(target_dir) + if launcher_path is not None and not launcher_path.exists(): + launcher_path = None + + rc_path = _terraformrc_path() + hosts: list[str] = [] + if rc_path.exists(): + try: + text = rc_path.read_text(encoding="utf-8") + except OSError: + text = "" + match = terraformrc.find_block(text) + if match is not None and match.group("name") == terraformrc.HELPER_NAME: + hosts.append("all Cloudsmith Terraform registries (credentials_helper)") + + return { + "launcher": str(launcher_path) if launcher_path is not None else None, + "hosts": hosts, + } diff --git a/cloudsmith_cli/credential_helpers/terraform/runtime.py b/cloudsmith_cli/credential_helpers/terraform/runtime.py new file mode 100644 index 00000000..58abc65a --- /dev/null +++ b/cloudsmith_cli/credential_helpers/terraform/runtime.py @@ -0,0 +1,188 @@ +# Copyright 2026 Cloudsmith Ltd +""" +Terraform credentials helper runtime. + +Transport-light protocol logic for Terraform's credentials-helper protocol. +This module is intentionally free of Click/sys imports so it can be unit-tested +without invoking the CLI machinery. + +Terraform runs a credentials helper once per credentials request it cannot +satisfy from a ``credentials`` block in the CLI configuration, invoking it as:: + + terraform-credentials-cloudsmith [args...] + +The current verbs are ``get``, ``store`` and ``forget``. This helper only +serves ``get``: + +* ``get`` for a Cloudsmith host with an available credential prints a JSON + credentials object (``{"token": "..."}``) to stdout and exits zero. +* ``get`` for a host we definitively have no credentials for (a non-Cloudsmith + host) prints an empty JSON object (``{}``) to stdout and exits zero, so + Terraform falls back to its own credential sources. +* ``get`` for a Cloudsmith host we cannot authenticate prints an + end-user-oriented error to stderr and exits non-zero. +* ``store``/``forget`` and any unknown verb print an error to stderr and exit + non-zero — credentials come from the Cloudsmith CLI's own provider chain, so + there is nothing to store or forget. + +See: https://developer.hashicorp.com/terraform/internals/credentials-helpers +""" + +import json +import logging + +from ..backends import BackendKind +from ..common import is_cloudsmith_domain, is_standard_cloudsmith_domain + +logger = logging.getLogger(__name__) + +_REFUSAL_MESSAGE = ( + "Error: Unable to retrieve credentials. " + "Provide credentials via the CLOUDSMITH_API_KEY environment variable, " + "credentials.ini, the system keyring, or an OIDC service. " + "Verify current authentication with `cloudsmith whoami --verbose`." +) + +_MISSING_ORG_MESSAGE = ( + "Error: No organisation configured. " + "Provide the Cloudsmith organisation via the --org flag, the " + "CLOUDSMITH_ORG environment variable, or the 'org' key in config.ini so " + "the token can be scoped as '{org}/{repo}/{token}'." +) + + +def get_token(hostname, credential=None, api_host=None, org=None): + """ + Get the token for a Cloudsmith Terraform registry. + + Verifies the hostname is a Cloudsmith registry (including custom domains) + and returns the token if one is available. + + Args: + hostname: The Terraform registry hostname + credential: Pre-resolved CredentialResult from the provider chain + api_host: Cloudsmith API host URL + org: Organisation slug whose custom domains to match against + + Returns: + str: The token Terraform sends as its bearer credential, or None when + this is not a Cloudsmith host or no credential is available. + """ + if not is_cloudsmith_domain( + hostname, + credential=credential, + api_host=api_host, + backend_kind=BackendKind.TERRAFORM, + org=org, + ): + return None + + if not credential or not credential.api_key: + return None + + return credential.api_key + + +def _execute_get( + hostname, credential, api_host, org, repo +) -> tuple[int, str | None, str | None]: + """Handle the 'get' verb of the Terraform credentials-helper protocol. + + Distinguishes two "no token" outcomes, per the protocol: + + * A host that is not a Cloudsmith one is *not ours to answer* — emit an + empty ``{}`` and exit zero so Terraform falls through to its own + credential sources rather than failing the request. + * A Cloudsmith host we hold no credential for is a *definitive failure* — + emit an actionable error on stderr and exit non-zero. + """ + if not hostname: + return (1, None, "Error: No hostname provided") + + if not is_cloudsmith_domain( + hostname, + credential=credential, + api_host=api_host, + backend_kind=BackendKind.TERRAFORM, + org=org, + ): + # Not a Cloudsmith host: emit an empty object so Terraform moves on to + # its own credential sources instead of treating this as an error. + return (0, "{}", None) + + # Standard *.cloudsmith.io/.com hosts carry the org in the token; a custom + # domain is already bound to a single organisation, so its token must omit + # the org and be scoped as "{repo}/{token}" instead. The org is therefore + # only required on the standard-domain path (checked after the domain match + # so foreign hosts still fall back cleanly via the empty object above). + standard_domain = is_standard_cloudsmith_domain(hostname) + if standard_domain and not org: + return (1, None, _MISSING_ORG_MESSAGE) + + token = get_token(hostname, credential=credential, api_host=api_host, org=org) + if not token: + return (1, None, _REFUSAL_MESSAGE) + + # Scope the token to a single repository. Standard domains also carry the + # org ("{org}/{repo}/{token}"); custom domains are already org-bound, so + # they omit it ("{repo}/{token}"). + if standard_domain: + scoped_token = f"{org}/{repo}/{token}" + else: + scoped_token = f"{repo}/{token}" + return (0, json.dumps({"token": scoped_token}), None) + + +def execute( + verb, hostname, credential=None, api_host=None, org=None, repo=None +) -> tuple[int, str | None, str | None]: + """ + Execute a Terraform credentials-helper protocol verb. + + Args: + verb: One of 'get', 'store', 'forget' + hostname: The registry hostname the verb applies to (used by 'get') + credential: Pre-resolved CredentialResult from the provider chain + api_host: Cloudsmith API host URL + org: Organisation slug whose custom domains to match against + + Returns: + A (exit_code, stdout_text, stderr_text) tuple. Either text value may + be None if there is nothing to write to that stream. + """ + if verb == "get": + try: + return _execute_get(hostname, credential, api_host, org, repo) + except Exception as exc: # pylint: disable=broad-except + # Protocol boundary: a credentials helper must never crash + # `terraform init` with a traceback. Covers network/SDK errors + # from the custom-domain lookup and TypeError from json.dumps — all + # degrade to a clean refusal (exit 1), not a traceback. + # (Exception does not catch KeyboardInterrupt/SystemExit, which is + # correct.) + logger.debug( + "terraform credentials-helper get failed: %s", exc, exc_info=True + ) + return (1, None, _REFUSAL_MESSAGE) + + if verb in ("store", "forget"): + # Credentials are resolved from the Cloudsmith CLI's own provider chain + # (API key, credentials.ini, keyring, OIDC), so there is nothing for + # Terraform to store or forget here. + return ( + 1, + None, + ( + f"Error: '{verb}' is not supported. Credentials are managed by " + "the Cloudsmith credential chain and cannot be stored or " + "forgotten by this helper." + ), + ) + + # Forward-compatibility: react to any unsupported verb with an error and a + # non-zero exit, as the protocol requires. + return ( + 1, + None, + f"Error: Unknown verb '{verb}'. Valid verbs: get, store, forget", + ) diff --git a/cloudsmith_cli/credential_helpers/terraform/terraformrc.py b/cloudsmith_cli/credential_helpers/terraform/terraformrc.py new file mode 100644 index 00000000..089c056b --- /dev/null +++ b/cloudsmith_cli/credential_helpers/terraform/terraformrc.py @@ -0,0 +1,133 @@ +# Copyright 2026 Cloudsmith Ltd +"""Idempotent management of the ``credentials_helper`` block in ``~/.terraformrc``. + +Terraform's CLI configuration file is HCL, but the one block this helper owns is +simple and regular enough to manage as text without pulling in an HCL parser +(which would also lose the user's comments and formatting on round-trip). The +functions here locate a ``credentials_helper "cloudsmith" { ... }`` block by +regex and add, replace, or remove *only* that block, leaving the rest of the +file byte-for-byte intact. + +Terraform allows at most one ``credentials_helper`` block in the whole file, so +if a block for a *different* helper is already present these helpers refuse +rather than silently producing an invalid config. + +This module is intentionally free of Click/sys imports so it can be unit-tested +without invoking the CLI machinery. +""" + +from __future__ import annotations + +import re + +HELPER_NAME = "cloudsmith" + +# Matches a whole `credentials_helper "" { ... }` block, capturing the +# helper name. The body is matched non-greedily up to the first closing brace on +# its own — the block this helper writes never contains nested braces, and a +# hand-written one would not either, so a flat match is sufficient and avoids a +# full HCL parser. DOTALL lets the body span lines. +_BLOCK_RE = re.compile( + r'credentials_helper\s+"(?P[^"]+)"\s*\{.*?\}', + re.DOTALL, +) + + +class TerraformrcConflictError(Exception): + """Raised when a credentials_helper block for a different helper exists. + + Terraform permits only one ``credentials_helper`` block, so we must not add + a second one, and we must not overwrite someone else's. + """ + + def __init__(self, existing_name: str) -> None: + self.existing_name = existing_name + super().__init__( + "~/.terraformrc already configures a different credentials helper " + f"({existing_name!r}). Terraform allows only one credentials_helper " + "block; remove the existing one before installing the Cloudsmith " + "helper." + ) + + +def render_block(args: list[str] | tuple[str, ...] = ()) -> str: + """Render the ``credentials_helper "cloudsmith"`` block for *args*. + + Args: + args: The values for the block's ``args = [...]`` list, e.g. + ``["--org", "acme", "-P", "ci"]``. + + Returns: + The HCL block as text, without a trailing newline. + """ + rendered_args = ", ".join(f'"{a}"' for a in args) + return f'credentials_helper "{HELPER_NAME}" {{\n args = [{rendered_args}]\n}}' + + +def find_block(text: str) -> re.Match[str] | None: + """Return the first ``credentials_helper`` block match in *text*, or None.""" + return _BLOCK_RE.search(text) + + +def add_or_update_block( + text: str, args: list[str] | tuple[str, ...] = () +) -> tuple[str, bool]: + """Return *text* with the Cloudsmith credentials_helper block installed. + + If a Cloudsmith block already exists it is replaced (so re-running with + different ``args`` updates it); if a block for a different helper exists a + :class:`TerraformrcConflictError` is raised; otherwise the block is appended. + + Args: + text: Current terraformrc content ("" when the file does not exist). + args: Values for the block's ``args`` list. + + Returns: + A ``(new_text, changed)`` tuple; *changed* is False when the file + already contained exactly the desired block. + """ + block = render_block(args) + match = find_block(text) + + if match is None: + # No credentials_helper at all: append our block, keeping any existing + # content and separating with a blank line. + if text.strip() == "": + new_text = block + "\n" + else: + separator = "" if text.endswith("\n") else "\n" + new_text = f"{text}{separator}\n{block}\n" + return new_text, new_text != text + + if match.group("name") != HELPER_NAME: + raise TerraformrcConflictError(match.group("name")) + + # Replace the existing Cloudsmith block in place. + new_text = text[: match.start()] + block + text[match.end() :] + return new_text, new_text != text + + +def remove_block(text: str) -> tuple[str, bool]: + """Return *text* with the Cloudsmith credentials_helper block removed. + + A block for a different helper is left untouched (nothing to remove). + + Args: + text: Current terraformrc content. + + Returns: + A ``(new_text, changed)`` tuple; *changed* is False when there was no + Cloudsmith block to remove. + """ + match = find_block(text) + if match is None or match.group("name") != HELPER_NAME: + return text, False + + # Drop the block and collapse the surrounding blank lines it leaves behind + # so we don't accumulate whitespace across install/uninstall cycles. + new_text = text[: match.start()] + text[match.end() :] + new_text = re.sub(r"\n{3,}", "\n\n", new_text) + new_text = new_text.strip("\n") + if new_text: + new_text += "\n" + return new_text, new_text != text