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..113f9366 100644 --- a/cloudsmith_cli/cli/tests/commands/test_credential_helper.py +++ b/cloudsmith_cli/cli/tests/commands/test_credential_helper.py @@ -9,6 +9,7 @@ import pytest from ....cli.commands.credential_helper.docker import docker +from ....core.api.init import unset_api_key from ....core.credentials.models import CredentialResult from ....credential_helpers.backends import BackendKind from ....credential_helpers.custom_domains import ( @@ -157,8 +158,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 +283,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 +303,40 @@ 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", + ) + + # initialise_api never clears the class-default api_key dict, so an + # X-Api-Key set by an earlier test would otherwise leak into this request. + unset_api_key() + 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 +438,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"] @@ -522,7 +562,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 +573,15 @@ def test_is_cloudsmith_domain( assert result is expected +def test_is_cloudsmith_domain_custom_domain_without_credential(monkeypatch): + """A custom-domain check without a credential refuses without an API call.""" + from ....credential_helpers.common import is_cloudsmith_domain + + monkeypatch.setenv("CLOUDSMITH_ORG", "acme") + + assert is_cloudsmith_domain("docker.acme.com", api_host=API_HOST) is False + + # --------------------------------------------------------------------------- # 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..d44289ca 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,7 @@ def test_docker_installer_status_type_contract(tmp_path, monkeypatch): "discovery_on", "no_discover", "missing_org", - "missing_api_key", + "missing_credential", "discovery_raises", ], ) @@ -347,19 +348,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 +378,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"): called = [] def _should_not_be_called(*_a, **_kw): @@ -388,11 +395,11 @@ 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) + cred = credential if scenario == "missing_org" else None + installer.install(bin_dir=str(bin_dir), discover=True, org=org, credential=cred) 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" @@ -405,7 +412,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 +471,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 +523,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/credential_helpers/common.py b/cloudsmith_cli/credential_helpers/common.py index f7bcf85b..a455b5d0 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: 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..cdbd7791 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,7 @@ def install( # --- Custom-domain auto-discovery (best-effort) --- if discover: - if org and api_key: + if org and credential: # 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 +142,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, ):