fix: stop recurring macOS keychain prompts and self-heal dead SSO sessions

cloudsmith_cli/cli/commands/auth.py

@@ -17,7 +17,12 @@
 
 
 def _perform_saml_authentication(
-    opts, owner, enable_token_creation=False, use_stderr=False, no_browser=False
+    opts,
+    owner,
+    enable_token_creation=False,
+    use_stderr=False,
+    no_browser=False,
+    profile=None,
 ):
     """Perform SAML authentication via web browser and local web server."""
     session = create_configured_session(opts)
@@ -63,6 +68,7 @@ def _perform_saml_authentication(
         debug=opts.debug,
         refresh_api_on_success=enable_token_creation,
         api_opts=opts.api_config,
+        profile=profile,
     )
 
     auth_server.handle_request()
@@ -187,6 +193,7 @@ def authenticate(
             enable_token_creation=enable_token_creation,
             use_stderr=use_stderr,
             no_browser=no_browser,
+            profile=ctx.meta.get("profile"),
         )
 
     if request_api_key_flag:

cloudsmith_cli/cli/commands/logout.py

@@ -33,8 +33,13 @@ def _clear_credentials(dry_run, use_stderr):
     return {"action": action, "files": list(creds_files)}
 
 
-def _clear_keyring(api_host, dry_run, use_stderr):
-    """Clear SSO tokens from keyring. Returns result dict."""
+def _clear_keyring(api_host, dry_run, use_stderr, profile=None):
+    """Clear SSO tokens from keyring. Returns result dict.
+
+    Remove the profile's own entries. The legacy unscoped entries hold
+    the default profile's session, so remove them only when the profile
+    has no entries of its own.
+    """
     if not keyring.should_use_keyring():
         click.secho(
             "Keyring is disabled (CLOUDSMITH_NO_KEYRING is set).",
@@ -43,15 +48,17 @@ def _clear_keyring(api_host, dry_run, use_stderr):
         )
         return {"action": "disabled"}
 
-    if not keyring.has_sso_tokens(api_host):
+    if not keyring.has_sso_tokens(api_host, profile=profile):
         click.echo("No SSO tokens found in system keyring.", err=use_stderr)
         return {"action": "not_found"}
 
     if dry_run:
         click.echo("Would remove SSO tokens from system keyring.", err=use_stderr)
         return {"action": "would_remove"}
 
-    deleted = keyring.delete_sso_tokens(api_host)
+    deleted = keyring.delete_sso_tokens(api_host, profile=profile, include_legacy=False)
+    if not deleted:
+        deleted = keyring.delete_sso_tokens(api_host)
     action = "removed" if deleted else "failed"
     msg = f"{'Removed' if deleted else 'Failed to remove'} SSO tokens from system keyring."
     click.secho(msg, fg=None if deleted else "red", err=use_stderr)
@@ -131,7 +138,7 @@ def logout(ctx, opts, api_host, keyring_only, config_only, dry_run):
         else {"action": "skipped", "files": []}
     )
     keyring_result = (
-        _clear_keyring(api_host, dry_run, use_stderr)
+        _clear_keyring(api_host, dry_run, use_stderr, profile=ctx.meta.get("profile"))
         if not config_only
         else {"action": "skipped"}
     )

cloudsmith_cli/cli/commands/whoami.py

@@ -36,11 +36,15 @@ def _get_api_key_source(opts):
     return {"configured": False, "source": None, "source_key": None}
 
 
-def _get_sso_status(api_host):
+def _get_sso_status(api_host, profile=None):
     """Return SSO token status from the system keyring."""
     enabled = keyring.should_use_keyring()
-    has_tokens = enabled and keyring.has_sso_tokens(api_host)
-    refreshed = keyring.get_refresh_attempted_at(api_host) if has_tokens else None
+    has_tokens = enabled and keyring.has_sso_tokens(api_host, profile=profile)
+    refreshed = (
+        keyring.get_refresh_attempted_at(api_host, profile=profile)
+        if has_tokens
+        else None
+    )
 
     return {
         "configured": has_tokens,
@@ -50,10 +54,10 @@ def _get_sso_status(api_host):
     }
 
 
-def _get_verbose_auth_data(opts, api_host):
+def _get_verbose_auth_data(opts, api_host, profile=None):
     """Gather all auth details for verbose output."""
     api_key_info = _get_api_key_source(opts)
-    sso_info = _get_sso_status(api_host)
+    sso_info = _get_sso_status(api_host, profile=profile)
 
     # Fetch token metadata (extra API call, graceful fallback)
     token_meta = None
@@ -171,7 +175,9 @@ def whoami(ctx, opts):
 
     if opts.verbose:
         api_host = getattr(opts.api_config, "host", None) or opts.api_host
-        data["auth"] = _get_verbose_auth_data(opts, api_host)
+        data["auth"] = _get_verbose_auth_data(
+            opts, api_host, profile=ctx.meta.get("profile")
+        )
 
     if utils.maybe_print_as_json(opts, data):
         if not is_auth:

cloudsmith_cli/cli/saml.py

@@ -2,7 +2,15 @@
 
 import requests
 
-from ..core.api.exceptions import ApiException
+from ..core.sso import raise_for_api_error, refresh_access_token
+
+__all__ = [
+    "create_configured_session",
+    "exchange_2fa_token",
+    "get_idp_url",
+    "raise_for_api_error",
+    "refresh_access_token",
+]
 
 
 def create_configured_session(opts):
@@ -26,28 +34,6 @@ def create_configured_session(opts):
     return session
 
 
-def raise_for_api_error(response):
-    """Raise :class:`ApiException` if *response* failed, keeping the API's detail.
-
-    Without the detail the exception renders as bare status text, so a caller
-    reporting it tells the user that something failed but never what.
-    """
-    try:
-        response.raise_for_status()
-    except requests.RequestException as exc:
-        try:
-            body = exc.response.json()
-        except ValueError:
-            body = None
-
-        raise ApiException(
-            response.status_code,
-            detail=body.get("detail") if isinstance(body, dict) else None,
-            headers=exc.response.headers,
-            body=exc.response.content,
-        )
-
-
 def get_idp_url(api_host, owner, session):
     org_saml_url = "{api_host}/orgs/{owner}/saml/?{params}".format(
         api_host=api_host,
@@ -82,25 +68,3 @@ def exchange_2fa_token(api_host, two_factor_token, totp_token, session):
     refresh_token = exchange_data.get("refresh_token")
 
     return (access_token, refresh_token)
-
-
-def refresh_access_token(api_host, access_token, refresh_token, session):
-    data = {"refresh_token": refresh_token}
-    url = f"{api_host}/user/refresh-token/"
-
-    headers = {"Authorization": f"Bearer {access_token}"}
-
-    response = session.post(
-        url,
-        data=data,
-        headers=headers,
-        timeout=30,
-    )
-
-    raise_for_api_error(response)
-
-    response_data = response.json()
-    access_token = response_data.get("access_token")
-    refresh_token = response_data.get("refresh_token")
-
-    return (access_token, refresh_token)

cloudsmith_cli/cli/tests/commands/test_logout.py

@@ -1,6 +1,6 @@
 import json
 import os
-from unittest.mock import patch
+from unittest.mock import call, patch
 
 import click.testing
 import pytest
@@ -18,10 +18,11 @@ def runner():
 
 @pytest.fixture
 def mock_no_keyring_env():
-    """Ensure CLOUDSMITH_NO_KEYRING and CLOUDSMITH_API_KEY are not set."""
+    """Ensure CLOUDSMITH_NO_KEYRING, CLOUDSMITH_API_KEY and CLOUDSMITH_PROFILE are not set."""
     env = os.environ.copy()
     env.pop("CLOUDSMITH_NO_KEYRING", None)
     env.pop("CLOUDSMITH_API_KEY", None)
+    env.pop("CLOUDSMITH_PROFILE", None)
     with patch.dict(os.environ, env, clear=True):
         yield
 
@@ -49,7 +50,9 @@ def test_full_logout(self, runner, mock_deps):
 
         assert result.exit_code == 0
         mock_creds.clear_api_key.assert_called_once_with(CREDS_PATH)
-        mock_keyring.delete_sso_tokens.assert_called_once_with(HOST)
+        mock_keyring.delete_sso_tokens.assert_called_once_with(
+            HOST, profile=None, include_legacy=False
+        )
         assert "Removed credentials from:" in result.output
         assert "Removed SSO tokens from system keyring" in result.output
 
@@ -59,7 +62,35 @@ def test_misconfigured_api_host_is_normalized(self, runner, mock_deps):
         result = runner.invoke(logout, ["--api-host", " api.example.com/ "])
 
         assert result.exit_code == 0
-        mock_keyring.delete_sso_tokens.assert_called_once_with(HOST)
+        mock_keyring.delete_sso_tokens.assert_called_once_with(
+            HOST, profile=None, include_legacy=False
+        )
+
+    def test_profile_logout_keeps_legacy_entries(self, runner, mock_deps):
+        """A scoped-profile logout does not touch the unscoped entries."""
+        _, mock_keyring = mock_deps
+        mock_keyring.delete_sso_tokens.return_value = True
+
+        result = runner.invoke(logout, ["--profile", "staging", "--api-host", HOST])
+
+        assert result.exit_code == 0
+        mock_keyring.delete_sso_tokens.assert_called_once_with(
+            HOST, profile="staging", include_legacy=False
+        )
+
+    def test_profile_logout_falls_back_to_legacy_entries(self, runner, mock_deps):
+        """When the profile has no scoped entries, remove the legacy entries."""
+        _, mock_keyring = mock_deps
+        mock_keyring.delete_sso_tokens.side_effect = [False, True]
+
+        result = runner.invoke(logout, ["--profile", "staging", "--api-host", HOST])
+
+        assert result.exit_code == 0
+        assert mock_keyring.delete_sso_tokens.call_args_list == [
+            call(HOST, profile="staging", include_legacy=False),
+            call(HOST),
+        ]
+        assert "Removed SSO tokens from system keyring" in result.output
 
     def test_dry_run(self, runner, mock_deps):
         mock_creds, mock_keyring = mock_deps

cloudsmith_cli/cli/tests/test_credential_helper_cargo.py

@@ -114,7 +114,7 @@ def test_get_returns_token_for_cloudsmith_registry(credential):
     assert messages[1] == {
         "Ok": {
             "kind": "get",
-            "token": "k_abc",
+            "token": "token k_abc",
             "cache": "session",
             "operation_independent": True,
         }
@@ -127,7 +127,7 @@ def test_get_serves_every_operation(operation, credential):
     request = _request(operation=operation, name="sample", vers="0.1.0")
     _, _, messages = _session(request, credential=credential)
 
-    assert messages[1]["Ok"]["token"] == "k_abc"
+    assert messages[1]["Ok"]["token"] == "token k_abc"
 
 
 def test_get_uses_the_cargo_backend_kind_for_custom_domains(credential):
@@ -149,10 +149,24 @@ def test_get_uses_the_cargo_backend_kind_for_custom_domains(credential):
 def test_index_url_keeps_its_sparse_prefix_out_of_the_host_match(credential):
     """Cargo's `sparse+` prefix and the repo path don't defeat the host check."""
     assert (
-        get_credentials(CLOUDSMITH_INDEX, credential=credential, org="acme") == "k_abc"
+        get_credentials(CLOUDSMITH_INDEX, credential=credential, org="acme")
+        == "token k_abc"
     )
 
 
+def test_get_returns_bearer_scheme_for_sso_credential():
+    """An SSO credential is returned with the Bearer authorization scheme."""
+    credential = CredentialResult(
+        api_key="jwt_token",
+        source_name="keyring",
+        auth_type="bearer",
+    )
+
+    _, _, messages = _session(_request(), credential=credential)
+
+    assert messages[1]["Ok"]["token"] == "Bearer jwt_token"
+
+
 # ---------------------------------------------------------------------------
 # 3. get — the refusal paths, which Cargo distinguishes
 # ---------------------------------------------------------------------------
@@ -247,7 +261,7 @@ def test_malformed_request_lines_are_answered_not_crashed(line, credential):
     assert (code, stderr) == (0, None)
     assert messages[1]["Err"]["kind"] == "other"
     # The following well-formed request is still served.
-    assert messages[2]["Ok"]["token"] == "k_abc"
+    assert messages[2]["Ok"]["token"] == "token k_abc"
 
 
 def test_blank_lines_are_skipped(credential):
@@ -322,7 +336,7 @@ def test_cli_speaks_the_protocol_on_stdin_and_stdout(runner):
     messages = [json.loads(line) for line in result.stdout.splitlines()]
     assert result.exit_code == 0
     assert messages[0] == hello()
-    assert messages[1]["Ok"]["token"] == "k_abc"
+    assert messages[1]["Ok"]["token"] == "token k_abc"
 
 
 def test_cli_accepts_the_cargo_plugin_flag_and_extra_provider_args(runner):
@@ -336,7 +350,7 @@ def test_cli_accepts_the_cargo_plugin_flag_and_extra_provider_args(runner):
 
     messages = [json.loads(line) for line in result.stdout.splitlines()]
     assert result.exit_code == 0
-    assert messages[1]["Ok"]["token"] == "k_abc"
+    assert messages[1]["Ok"]["token"] == "token k_abc"
 
 
 def test_cli_exits_non_zero_with_a_hint_when_no_credential_resolves(runner):

cloudsmith_cli/cli/tests/test_webserver.py

@@ -64,6 +64,7 @@ def mock_handler(self):
             )
             handler.server_instance = MagicMock()
             handler.server_instance.api_host = "https://api.cloudsmith.io"
+            handler.server_instance.profile = None
             handler.refresh_api_on_success = False
             handler.session = MagicMock()
             handler.debug = False
@@ -99,6 +100,41 @@ def test_store_sso_tokens_called_when_keyring_enabled(self, mock_handler):
                 "https://api.cloudsmith.io",
                 "test_access_token",
                 "test_refresh_token",
+                profile=None,
+            )
+
+    def test_store_sso_tokens_receives_profile(self, mock_handler):
+        """Verify store_sso_tokens receives the profile from the server."""
+        mock_handler.server_instance.profile = "staging"
+        with (
+            patch(
+                "cloudsmith_cli.cli.webserver.store_sso_tokens", return_value=True
+            ) as mock_store,
+            patch.object(mock_handler, "_return_success_response"),
+            patch.object(
+                AuthenticationWebRequestHandler,
+                "query_data",
+                new_callable=PropertyMock,
+            ) as mock_query,
+            patch.object(
+                AuthenticationWebRequestHandler,
+                "api_host",
+                new_callable=PropertyMock,
+            ) as mock_host,
+        ):
+            mock_query.return_value = {
+                "access_token": "test_access_token",
+                "refresh_token": "test_refresh_token",
+            }
+            mock_host.return_value = "https://api.cloudsmith.io"
+
+            mock_handler.do_GET()
+
+            mock_store.assert_called_once_with(
+                "https://api.cloudsmith.io",
+                "test_access_token",
+                "test_refresh_token",
+                profile="staging",
             )
 
     def test_message_shown_when_keyring_disabled(self, mock_handler):