refactor(no-ticket): pass resolved credentials through credential-helper plumbing

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:

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,14 +564,53 @@ 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
 
     result = is_cloudsmith_domain(host, **kwargs)
     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
 # ---------------------------------------------------------------------------

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
 # ---------------------------------------------------------------------------

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")