diff --git a/cloudsmith_cli/cli/commands/credential_helper/manage.py b/cloudsmith_cli/cli/commands/credential_helper/manage.py index 54d3a3ae..2dd4b395 100644 --- a/cloudsmith_cli/cli/commands/credential_helper/manage.py +++ b/cloudsmith_cli/cli/commands/credential_helper/manage.py @@ -139,12 +139,6 @@ def install_cmd( """ installer = _get_installer(helper) org = org or os.environ.get("CLOUDSMITH_ORG", "").strip() or None - api_key = opts.credential.api_key if opts.credential else None - auth_type = ( - getattr(opts.credential, "auth_type", "api_key") - if opts.credential - else "api_key" - ) try: actions = installer.install( bin_dir=bin_dir, @@ -153,8 +147,7 @@ def install_cmd( discover=not no_discover, refresh=refresh, org=org, - api_key=api_key, - auth_type=auth_type, + credential=opts.credential, api_host=opts.api_host, ) except OSError as exc: diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py index 98cf0a5a..17a15dad 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py @@ -9,8 +9,10 @@ import pytest from ....cli.commands.credential_helper.docker import docker +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.custom_domains import ( CustomDomain, get_cache_path, @@ -157,8 +159,9 @@ def test_get_credentials(server_url, credential, is_cloudsmith_return, expected) with patch( "cloudsmith_cli.credential_helpers.docker.runtime.is_cloudsmith_domain", return_value=is_cloudsmith_return, - ): + ) as mock_check: result = helper_get_credentials(server_url, credential=credential) + assert mock_check.call_args.kwargs["credential"] is credential assert result == expected @@ -281,7 +284,8 @@ def test_get_custom_domains_status_matrix( content_type="application/json", ) - result = get_custom_domains("acme", api_key="k_abc", api_host=API_HOST) + credential = CredentialResult(api_key="k_abc", source_name="test") + result = get_custom_domains("acme", credential=credential, api_host=API_HOST) cache = read_cache(get_cache_path("acme")) if expect_domains: @@ -300,6 +304,43 @@ def test_get_custom_domains_status_matrix( assert httpretty.last_request().headers.get("X-Api-Key") == "k_abc" +@httpretty.activate(allow_net_connect=False) +def test_get_custom_domains_bearer_credential_sends_authorization_header( + tmp_path, monkeypatch +): + """A bearer credential authenticates the lookup with its own header scheme. + + The credential object flows through to the API layer unmodified, so an + SSO access token goes out as ``Authorization: Bearer``, never ``X-Api-Key``. + """ + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + httpretty.register_uri( + httpretty.GET, + f"{API_HOST}/orgs/acme/custom-domains/", + body=json.dumps([]), + status=200, + content_type="application/json", + ) + + # A previously configured API key must not leak into a bearer-authenticated + # request: initialise_api clears the class-default X-Api-Key when switching. + initialise_api( + host=API_HOST, + credential=CredentialResult(api_key="k_stale", source_name="test"), + ) + credential = CredentialResult( + api_key="jwt_token", source_name="test", auth_type="bearer" + ) + get_custom_domains("acme", credential=credential, api_host=API_HOST) + + headers = httpretty.last_request().headers + assert headers.get("Authorization") == "Bearer jwt_token" + assert headers.get("X-Api-Key") is None + + # --------------------------------------------------------------------------- # 7. get_custom_domains — cache edge cases # --------------------------------------------------------------------------- @@ -401,7 +442,10 @@ def test_get_format_domains_filters_correctly(tmp_path, monkeypatch): ) hosts = get_format_domains( - "acme", BackendKind.DOCKER, api_key="k", api_host=API_HOST + "acme", + BackendKind.DOCKER, + credential=CredentialResult(api_key="k", source_name="test"), + api_host=API_HOST, ) assert hosts == ["docker.acme.com"] @@ -507,8 +551,6 @@ def test_is_cloudsmith_domain( tmp_path, monkeypatch, host, env_org, cached_domains, backend_kind, expected ): """is_cloudsmith_domain returns correct bool for standard, custom, and edge cases.""" - from ....credential_helpers.common import is_cloudsmith_domain - monkeypatch.setattr( "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", lambda: str(tmp_path), @@ -522,7 +564,10 @@ def test_is_cloudsmith_domain( if cached_domains is not None: write_cache(get_cache_path(env_org), cached_domains) - kwargs = {"api_key": "k_abc", "api_host": API_HOST} + kwargs = { + "credential": CredentialResult(api_key="k_abc", source_name="test"), + "api_host": API_HOST, + } if backend_kind is not None: kwargs["backend_kind"] = backend_kind @@ -530,6 +575,42 @@ def test_is_cloudsmith_domain( assert result is expected +@pytest.mark.parametrize( + "credential", + [ + # no credential at all + None, + # a credential carrying no usable key — truthy as an object, so the + # guard has to look at api_key, not just the credential + CredentialResult(api_key="", source_name="test"), + ], +) +@httpretty.activate(allow_net_connect=False) +def test_is_cloudsmith_domain_custom_domain_without_credential( + tmp_path, monkeypatch, credential +): + """A custom-domain check without a usable credential refuses without an API call.""" + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.get_default_config_path", + lambda: str(tmp_path), + ) + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + + called = [] + monkeypatch.setattr( + "cloudsmith_cli.credential_helpers.custom_domains.list_custom_domains", + lambda *_a, **_kw: called.append(True) or [], + ) + + assert ( + is_cloudsmith_domain( + "docker.acme.com", credential=credential, api_host=API_HOST + ) + is False + ) + assert not called, "the custom-domain API must not be queried without a credential" + + # --------------------------------------------------------------------------- # 10. Docker runtime backend_kind wiring # --------------------------------------------------------------------------- diff --git a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py index 09391769..cb43a5ba 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper_install.py @@ -13,6 +13,7 @@ import click.testing import pytest +from ....core.credentials.models import CredentialResult from ....credential_helpers.docker.installer import DockerInstaller from ....credential_helpers.launchers import ( _launcher_content, @@ -333,7 +334,9 @@ def test_docker_installer_status_type_contract(tmp_path, monkeypatch): "discovery_on", "no_discover", "missing_org", - "missing_api_key", + "missing_credential", + "blank_credential", + "dry_run", "discovery_raises", ], ) @@ -347,19 +350,25 @@ def test_autodiscovery(tmp_path, monkeypatch, scenario): bin_dir = tmp_path / "bin" monkeypatch.setenv("PATH", str(bin_dir)) + credential = CredentialResult(api_key="k_test", source_name="test") + if scenario == "discovery_on": - monkeypatch.setattr( - _INSTALLER_GET_FORMAT_DOMAINS, - lambda *_a, **_kw: ["docker.acme.com"], - ) + captured = {} + + def _fake_get_format_domains(*_a, **kwargs): + captured.update(kwargs) + return ["docker.acme.com"] + + monkeypatch.setattr(_INSTALLER_GET_FORMAT_DOMAINS, _fake_get_format_domains) installer = DockerInstaller() actions = installer.install( - bin_dir=str(bin_dir), discover=True, org="acme", api_key="k_test" + bin_dir=str(bin_dir), discover=True, org="acme", credential=credential ) cfg = json.loads((docker_dir / "config.json").read_text()) assert cfg["credHelpers"]["docker.cloudsmith.io"] == "cloudsmith" assert cfg["credHelpers"]["docker.acme.com"] == "cloudsmith" assert any("discovered" in a and "1" in a for a in actions) + assert captured["credential"] is credential elif scenario == "no_discover": called = [] @@ -371,14 +380,14 @@ def _should_not_be_called(*_a, **_kw): monkeypatch.setattr(_INSTALLER_GET_FORMAT_DOMAINS, _should_not_be_called) installer = DockerInstaller() installer.install( - bin_dir=str(bin_dir), discover=False, org="acme", api_key="k_test" + bin_dir=str(bin_dir), discover=False, org="acme", credential=credential ) assert not called, "get_format_domains must not be called when discover=False" cfg = json.loads((docker_dir / "config.json").read_text()) assert "docker.cloudsmith.io" in cfg["credHelpers"] assert "docker.acme.com" not in cfg["credHelpers"] - elif scenario in ("missing_org", "missing_api_key"): + elif scenario in ("missing_org", "missing_credential", "blank_credential"): called = [] def _should_not_be_called(*_a, **_kw): @@ -388,14 +397,40 @@ def _should_not_be_called(*_a, **_kw): monkeypatch.setattr(_INSTALLER_GET_FORMAT_DOMAINS, _should_not_be_called) installer = DockerInstaller() org = None if scenario == "missing_org" else "acme" - api_key = "k_test" if scenario == "missing_org" else None - installer.install(bin_dir=str(bin_dir), discover=True, org=org, api_key=api_key) + creds = { + "missing_org": credential, + "missing_credential": None, + "blank_credential": CredentialResult(api_key="", source_name="test"), + } + installer.install( + bin_dir=str(bin_dir), discover=True, org=org, credential=creds[scenario] + ) assert ( not called - ), "get_format_domains must not be called when org/api_key absent" + ), "get_format_domains must not be called when org/credential absent" cfg = json.loads((docker_dir / "config.json").read_text()) assert cfg["credHelpers"]["docker.cloudsmith.io"] == "cloudsmith" + elif scenario == "dry_run": + called = [] + + def _should_not_be_called(*_a, **_kw): + called.append(True) + return [] + + monkeypatch.setattr(_INSTALLER_GET_FORMAT_DOMAINS, _should_not_be_called) + installer = DockerInstaller() + actions = installer.install( + bin_dir=str(bin_dir), + discover=True, + org="acme", + credential=credential, + dry_run=True, + ) + assert not called, "a dry run must not query the custom-domain API" + assert not (docker_dir / "config.json").exists() + assert any("skipped custom-domain auto-discovery" in a for a in actions) + else: # discovery_raises — graceful failure guard def _raise(*_a, **_kw): @@ -405,7 +440,7 @@ def _raise(*_a, **_kw): installer = DockerInstaller() # Must NOT raise actions = installer.install( - bin_dir=str(bin_dir), discover=True, org="acme", api_key="k_test" + bin_dir=str(bin_dir), discover=True, org="acme", credential=credential ) cfg = json.loads((docker_dir / "config.json").read_text()) assert cfg["credHelpers"]["docker.cloudsmith.io"] == "cloudsmith" @@ -464,7 +499,11 @@ def _fake_list(*_a, **_kw): "cloudsmith_cli.credential_helpers.custom_domains.list_custom_domains", _fake_list, ): - result = get_custom_domains("acme", api_key="k", refresh=refresh) + result = get_custom_domains( + "acme", + credential=CredentialResult(api_key="k", source_name="test"), + refresh=refresh, + ) if refresh: # API must have been called @@ -512,6 +551,33 @@ def test_manage_cli_dry_run_exits_0(runner, tmp_path, monkeypatch): assert "would" in result.output.lower() or "dry run" in result.output.lower() +def test_manage_cli_passes_resolved_credential_to_installer( + runner, tmp_path, monkeypatch +): + """install hands the resolved CredentialResult to the installer intact.""" + monkeypatch.setenv("DOCKER_CONFIG", str(tmp_path / ".docker")) + + from ....cli.commands.credential_helper.manage import install_cmd + + with patch.object(DockerInstaller, "install", return_value=[]) as mock_install: + result = runner.invoke( + install_cmd, + [ + "docker", + "--no-discover", + "--bin-dir", + str(tmp_path / "bin"), + "--api-key", + "k_flag", + ], + ) + + assert result.exit_code == 0, result.output + credential = mock_install.call_args.kwargs["credential"] + assert isinstance(credential, CredentialResult) + assert credential.api_key == "k_flag" + + # --------------------------------------------------------------------------- # 14. PATH warning # --------------------------------------------------------------------------- diff --git a/cloudsmith_cli/core/api/init.py b/cloudsmith_cli/core/api/init.py index f1baf335..8a373635 100644 --- a/cloudsmith_cli/core/api/init.py +++ b/cloudsmith_cli/core/api/init.py @@ -43,6 +43,9 @@ def initialise_api( if credential: if credential.auth_type == "bearer": + # set_default() makes api_key sticky across calls, so an X-Api-Key + # left by an earlier credential would ride along with the bearer token. + config.api_key.pop("X-Api-Key", None) config.headers["Authorization"] = f"Bearer {credential.api_key}" if config.debug: click.echo("SSO access token config value set") diff --git a/cloudsmith_cli/credential_helpers/common.py b/cloudsmith_cli/credential_helpers/common.py index f7bcf85b..c3166948 100644 --- a/cloudsmith_cli/credential_helpers/common.py +++ b/cloudsmith_cli/credential_helpers/common.py @@ -48,9 +48,7 @@ def extract_hostname(url): return hostname -def is_cloudsmith_domain( - url, api_key=None, auth_type="api_key", api_host=None, backend_kind=None -): +def is_cloudsmith_domain(url, credential=None, api_host=None, backend_kind=None): """ Check if a URL points to a Cloudsmith service. @@ -59,8 +57,7 @@ def is_cloudsmith_domain( Args: url: URL or hostname to check - api_key: API key/token for authenticating custom domain lookups - auth_type: "api_key" (X-Api-Key header) or "bearer" (Authorization: Bearer) + credential: Resolved CredentialResult for authenticating custom domain lookups api_host: Cloudsmith API host URL backend_kind: If given, custom domains only match when their backend_kind equals it (standard *.cloudsmith.io domains always match regardless). @@ -86,7 +83,7 @@ def is_cloudsmith_domain( if not org: return False - if not api_key: + if not credential or not credential.api_key: return False if backend_kind is not None: @@ -95,17 +92,14 @@ def is_cloudsmith_domain( for host in get_format_domains( org, backend_kind, - api_key=api_key, - auth_type=auth_type, + credential=credential, api_host=api_host, ) } else: hosts = { d.host.lower() - for d in get_custom_domains( - org, api_key=api_key, auth_type=auth_type, api_host=api_host - ) + for d in get_custom_domains(org, credential=credential, api_host=api_host) if d.enabled and d.validated } return hostname in hosts diff --git a/cloudsmith_cli/credential_helpers/custom_domains.py b/cloudsmith_cli/credential_helpers/custom_domains.py index e854b2d0..5c92cf03 100644 --- a/cloudsmith_cli/credential_helpers/custom_domains.py +++ b/cloudsmith_cli/credential_helpers/custom_domains.py @@ -11,7 +11,6 @@ import time from dataclasses import dataclass from pathlib import Path -from typing import Literal from ..cli.config import get_default_config_path from ..core.api.exceptions import ApiException @@ -154,8 +153,7 @@ def write_cache(cache_path: Path, domains: list[CustomDomain]) -> None: def get_custom_domains( # pylint: disable=too-many-return-statements org: str, *, - api_key: str | None = None, - auth_type: str = "api_key", + credential: CredentialResult | None = None, api_host: str | None = None, refresh: bool = False, ) -> list[CustomDomain]: @@ -166,8 +164,8 @@ def get_custom_domains( # pylint: disable=too-many-return-statements Args: org: Organization slug - api_key: Optional API key/token for authentication - auth_type: "api_key" (uses X-Api-Key header) or "bearer" (uses Authorization: Bearer) + credential: Optional resolved credential for authentication; it carries + its own auth scheme (X-Api-Key vs Authorization: Bearer) api_host: Cloudsmith API host URL (including version). Taken from the SDK configuration default when not provided. refresh: When ``True``, skip the cache read and always fetch from the API. @@ -192,18 +190,6 @@ def get_custom_domains( # pylint: disable=too-many-return-statements logger.debug("Fetching custom domains from API for %s", org) - normalized_auth_type: Literal["api_key", "bearer"] = ( - "bearer" if auth_type == "bearer" else "api_key" - ) - credential = ( - CredentialResult( - api_key=api_key, - source_name="credential-helper", - auth_type=normalized_auth_type, - ) - if api_key - else None - ) initialise_api(host=api_host, credential=credential) try: @@ -252,8 +238,7 @@ def get_format_domains( org: str, backend_kind: int, *, - api_key: str | None = None, - auth_type: str = "api_key", + credential: CredentialResult | None = None, api_host: str | None = None, refresh: bool = False, ) -> list[str]: @@ -263,8 +248,7 @@ def get_format_domains( Args: org: Organization slug backend_kind: BackendKind int value (e.g. BackendKind.DOCKER == 6) - api_key: Optional API key/token for authentication - auth_type: "api_key" or "bearer" + credential: Optional resolved credential for authentication api_host: Cloudsmith API host URL refresh: When ``True``, bypass the cache and fetch fresh data from the API. @@ -272,7 +256,7 @@ def get_format_domains( List of hostnames that are enabled, validated, and match the given backend_kind. """ domains = get_custom_domains( - org, api_key=api_key, auth_type=auth_type, api_host=api_host, refresh=refresh + org, credential=credential, api_host=api_host, refresh=refresh ) return [ d.host diff --git a/cloudsmith_cli/credential_helpers/docker/installer.py b/cloudsmith_cli/credential_helpers/docker/installer.py index ea518f4d..5ad9356e 100644 --- a/cloudsmith_cli/credential_helpers/docker/installer.py +++ b/cloudsmith_cli/credential_helpers/docker/installer.py @@ -15,6 +15,7 @@ from pathlib import Path from ...core.cache_utils import merge_json_file +from ...core.credentials.models import CredentialResult from ..backends import BackendKind from ..custom_domains import get_format_domains from ..launchers import is_on_path, remove_launcher, resolve_bin_dir, write_launcher @@ -80,8 +81,7 @@ def install( discover: bool = True, refresh: bool = False, org: str | None = None, - api_key: str | None = None, - auth_type: str = "api_key", + credential: CredentialResult | None = None, api_host: str | None = None, dry_run: bool = False, ) -> list[str]: @@ -107,10 +107,8 @@ def install( the API. Only meaningful when *discover* is also ``True``. org: Cloudsmith organisation slug used for custom-domain discovery. - api_key: - API key used for custom-domain discovery. - auth_type: - Credential type: ``"api_key"`` (default) or ``"bearer"``. + credential: + Resolved credential used for custom-domain discovery. api_host: Cloudsmith API host URL override. dry_run: @@ -133,7 +131,11 @@ def install( # --- Custom-domain auto-discovery (best-effort) --- if discover: - if org and api_key: + if dry_run: + # Discovery queries the API and refreshes the on-disk domain + # cache, neither of which a "no changes" preview may do. + actions.append("skipped custom-domain auto-discovery (dry run)") + elif org and credential and credential.api_key: # Discovery boundary: network/SDK errors must never abort the # default install. ApiException is already handled inside # get_format_domains; this broad catch is the deliberate outer @@ -144,8 +146,7 @@ def install( discovered = get_format_domains( org, BackendKind.DOCKER, - api_key=api_key, - auth_type=auth_type, + credential=credential, api_host=api_host, refresh=refresh, ) diff --git a/cloudsmith_cli/credential_helpers/docker/runtime.py b/cloudsmith_cli/credential_helpers/docker/runtime.py index b078216c..828fabef 100644 --- a/cloudsmith_cli/credential_helpers/docker/runtime.py +++ b/cloudsmith_cli/credential_helpers/docker/runtime.py @@ -45,8 +45,7 @@ def get_credentials(server_url, credential=None, api_host=None): if not is_cloudsmith_domain( server_url, - api_key=credential.api_key, - auth_type=getattr(credential, "auth_type", "api_key"), + credential=credential, api_host=api_host, backend_kind=BackendKind.DOCKER, ):