From 7e3c0e33f4f15021418b6684687b7ce1b8c6e397 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Sun, 23 Aug 2026 12:38:57 +0100 Subject: [PATCH 1/7] fix(no-ticket): stop recurring macOS keychain prompts and self-heal dead SSO sessions The keyring library implements each keychain write as a delete followed by a re-create. The re-created item has a fresh access control list, so every "Always Allow" grant was lost on the next token refresh and the keychain prompts returned forever. - Add core/macos_keychain.py and update keychain items in place with SecItemUpdate, which keeps the access control list. Fall back to the normal keyring write when the item does not exist. Resolve the chainer backend to its first member before the update. - Scope keyring service names by profile. Non-default profiles read the legacy unscoped entries as a fallback, so existing sessions stay valid and migrate to scoped entries on the next refresh. The default profile keeps the unscoped names. - Clear a profile's SSO tokens when the server rejects the refresh (400/401/403/422), so the CLI returns to a clean logged-out state instead of retrying dead tokens every 30 minutes. Transient failures keep the throttled retry. Treat a refresh response without an access token as a failure, and skip the refresh when no refresh token is stored. Co-Authored-By: Claude Fable 5 --- cloudsmith_cli/cli/commands/auth.py | 9 +- cloudsmith_cli/cli/commands/logout.py | 8 +- cloudsmith_cli/cli/commands/whoami.py | 18 +- .../cli/tests/commands/test_logout.py | 5 +- cloudsmith_cli/cli/tests/test_webserver.py | 36 +++ cloudsmith_cli/cli/webserver.py | 54 ++-- .../credentials/providers/keyring_provider.py | 80 ++++- cloudsmith_cli/core/keyring.py | 141 ++++++-- cloudsmith_cli/core/macos_keychain.py | 168 ++++++++++ cloudsmith_cli/core/tests/test_keyring.py | 303 +++++++++++++++++- .../core/tests/test_keyring_provider.py | 211 +++++++++++- cloudsmith_cli/core/tests/test_metadata.py | 8 +- 12 files changed, 937 insertions(+), 104 deletions(-) create mode 100644 cloudsmith_cli/core/macos_keychain.py diff --git a/cloudsmith_cli/cli/commands/auth.py b/cloudsmith_cli/cli/commands/auth.py index 215c031a..0955b4c7 100644 --- a/cloudsmith_cli/cli/commands/auth.py +++ b/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: diff --git a/cloudsmith_cli/cli/commands/logout.py b/cloudsmith_cli/cli/commands/logout.py index 53e9d184..09a4b15e 100644 --- a/cloudsmith_cli/cli/commands/logout.py +++ b/cloudsmith_cli/cli/commands/logout.py @@ -33,7 +33,7 @@ def _clear_credentials(dry_run, use_stderr): return {"action": action, "files": list(creds_files)} -def _clear_keyring(api_host, dry_run, use_stderr): +def _clear_keyring(api_host, dry_run, use_stderr, profile=None): """Clear SSO tokens from keyring. Returns result dict.""" if not keyring.should_use_keyring(): click.secho( @@ -43,7 +43,7 @@ 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"} @@ -51,7 +51,7 @@ def _clear_keyring(api_host, dry_run, use_stderr): 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) 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) @@ -128,7 +128,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"} ) diff --git a/cloudsmith_cli/cli/commands/whoami.py b/cloudsmith_cli/cli/commands/whoami.py index 0b701095..81381be9 100644 --- a/cloudsmith_cli/cli/commands/whoami.py +++ b/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: diff --git a/cloudsmith_cli/cli/tests/commands/test_logout.py b/cloudsmith_cli/cli/tests/commands/test_logout.py index 87d2be45..102fee0f 100644 --- a/cloudsmith_cli/cli/tests/commands/test_logout.py +++ b/cloudsmith_cli/cli/tests/commands/test_logout.py @@ -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,7 @@ 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) assert "Removed credentials from:" in result.output assert "Removed SSO tokens from system keyring" in result.output diff --git a/cloudsmith_cli/cli/tests/test_webserver.py b/cloudsmith_cli/cli/tests/test_webserver.py index 66fb660c..e5aa53ec 100644 --- a/cloudsmith_cli/cli/tests/test_webserver.py +++ b/cloudsmith_cli/cli/tests/test_webserver.py @@ -61,6 +61,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 @@ -96,6 +97,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): diff --git a/cloudsmith_cli/cli/webserver.py b/cloudsmith_cli/cli/webserver.py index 3a2d2d3c..f9b847f2 100644 --- a/cloudsmith_cli/cli/webserver.py +++ b/cloudsmith_cli/cli/webserver.py @@ -33,6 +33,7 @@ def __init__( self.debug = kwargs.get("debug", False) self.refresh_api_on_success = kwargs.get("refresh_api_on_success", False) self.api_opts = kwargs.get("api_opts") + self.profile = kwargs.get("profile") self.sso_access_token = None self.exception = None @@ -132,6 +133,11 @@ def api_host(self): """Get the API host from the server instance.""" return self.server_instance.api_host if self.server_instance else None + @property + def profile(self): + """Get the profile from the server instance.""" + return self.server_instance.profile if self.server_instance else None + def _return_response(self, status=200, message=None): self.send_response(status) self.send_header("Content-Type", "text/html; charset=utf-8") @@ -176,6 +182,24 @@ def url(self): def query_data(self): return dict(parse_qsl(self.url.query)) + def _store_authentication_result(self, access_token, refresh_token): + # Store the access token on the server instance so it can be + # passed directly to initialise_api(), avoiding a keyring + # roundtrip (critical when CLOUDSMITH_NO_KEYRING is set). + if self.server_instance: + self.server_instance.sso_access_token = access_token + + if not store_sso_tokens( + self.api_host, access_token, refresh_token, profile=self.profile + ): + click.echo( + "SSO tokens not stored (CLOUDSMITH_NO_KEYRING is set)", + err=True, + ) + + if self.refresh_api_on_success and self.server_instance: + self.server_instance.refresh_api_config_after_auth() + def do_GET(self): access_token = self.query_data.get("access_token") refresh_token = self.query_data.get("refresh_token") @@ -191,21 +215,7 @@ def do_GET(self): try: if access_token: - # Store the access token on the server instance so it can be - # passed directly to initialise_api(), avoiding a keyring - # roundtrip (critical when CLOUDSMITH_NO_KEYRING is set). - if self.server_instance: - self.server_instance.sso_access_token = access_token - - if not store_sso_tokens(self.api_host, access_token, refresh_token): - click.echo( - "SSO tokens not stored (CLOUDSMITH_NO_KEYRING is set)", - err=True, - ) - - if self.refresh_api_on_success and self.server_instance: - self.server_instance.refresh_api_config_after_auth() - + self._store_authentication_result(access_token, refresh_token) self._return_success_response() return @@ -218,19 +228,7 @@ def do_GET(self): self.api_host, two_factor_token, totp_token, session=self.session ) - # Store the access token on the server instance (same as above) - if self.server_instance: - self.server_instance.sso_access_token = access_token - - if not store_sso_tokens(self.api_host, access_token, refresh_token): - click.echo( - "SSO tokens not stored (CLOUDSMITH_NO_KEYRING is set)", - err=True, - ) - - if self.refresh_api_on_success and self.server_instance: - self.server_instance.refresh_api_config_after_auth() - + self._store_authentication_result(access_token, refresh_token) click.secho("\nAuthentication complete", fg="green", err=True) self._return_success_response() return diff --git a/cloudsmith_cli/core/credentials/providers/keyring_provider.py b/cloudsmith_cli/core/credentials/providers/keyring_provider.py index 814a8e4b..1d76de21 100644 --- a/cloudsmith_cli/core/credentials/providers/keyring_provider.py +++ b/cloudsmith_cli/core/credentials/providers/keyring_provider.py @@ -6,11 +6,36 @@ from ....cli.saml import refresh_access_token from ....core import keyring +from ...api.exceptions import ApiException from ..models import CredentialContext, CredentialResult from ..provider import CredentialProvider logger = logging.getLogger(__name__) +REFRESH_REJECTED_STATUSES = (400, 401, 403, 422) + + +def _handle_refresh_failure(context, wipe_tokens): + """Record a refresh failure and clear rejected tokens. + + A definitive rejection means the stored tokens are dead. Remove the + profile's own entries so the CLI returns to a clean logged-out + state instead of retrying dead tokens on every command. When the + profile has no entries of its own, the rejected tokens came from + the legacy unscoped entries, so remove those. When no entry was + removed, stamp the attempt time to throttle the next refresh. + """ + tokens_removed = False + if wipe_tokens: + tokens_removed = keyring.delete_sso_tokens( + context.api_host, profile=context.profile, include_legacy=False + ) + if not tokens_removed: + tokens_removed = keyring.delete_sso_tokens(context.api_host) + if not tokens_removed: + keyring.update_refresh_attempted_at(context.api_host, profile=context.profile) + context.keyring_refresh_failed = True + class KeyringProvider(CredentialProvider): """Resolves credentials from SAML tokens stored in the system keyring.""" @@ -22,33 +47,54 @@ def resolve(self, context: CredentialContext) -> CredentialResult | None: return None api_host = context.api_host - access_token = keyring.get_access_token(api_host) + profile = context.profile + access_token = keyring.get_access_token(api_host, profile=profile) if not access_token: return None try: - if keyring.should_refresh_access_token(api_host): + if keyring.should_refresh_access_token(api_host, profile=profile): if not context.session: logger.debug( "Session unavailable; skipping token refresh, using existing token" ) else: - refresh_token = keyring.get_refresh_token(api_host) - new_access_token, new_refresh_token = refresh_access_token( - api_host, - access_token, - refresh_token, - session=context.session, - ) - keyring.store_sso_tokens( - api_host, new_access_token, new_refresh_token - ) - access_token = new_access_token - except Exception: # pylint: disable=broad-exception-caught - keyring.update_refresh_attempted_at(api_host) - context.keyring_refresh_failed = True - logger.debug("Failed to refresh SAML token", exc_info=True) + refresh_token = keyring.get_refresh_token(api_host, profile=profile) + if not refresh_token: + logger.debug( + "No refresh token stored; using the existing access token" + ) + else: + new_access_token, new_refresh_token = refresh_access_token( + api_host, + access_token, + refresh_token, + session=context.session, + ) + if not new_access_token: + logger.debug("The refresh response has no access token") + _handle_refresh_failure(context, wipe_tokens=False) + return None + keyring.store_sso_tokens( + api_host, + new_access_token, + new_refresh_token, + profile=profile, + ) + access_token = new_access_token + except Exception as exc: # pylint: disable=broad-exception-caught + wipe_tokens = ( + isinstance(exc, ApiException) + and exc.status in REFRESH_REJECTED_STATUSES + ) + if wipe_tokens: + logger.debug( + "SSO refresh rejected; clearing stored SSO tokens", exc_info=True + ) + else: + logger.debug("Failed to refresh SAML token", exc_info=True) + _handle_refresh_failure(context, wipe_tokens=wipe_tokens) return None return CredentialResult( diff --git a/cloudsmith_cli/core/keyring.py b/cloudsmith_cli/core/keyring.py index 5e721098..a3f0882e 100644 --- a/cloudsmith_cli/core/keyring.py +++ b/cloudsmith_cli/core/keyring.py @@ -1,5 +1,6 @@ import getpass import os +import sys from datetime import datetime, timedelta, timezone import keyring @@ -24,6 +25,30 @@ def _get_username(): return getpass.getuser() +def _is_scoped_profile(profile): + return bool(profile) and profile != "default" + + +def _format_key(template, api_host, profile): + key = template.format(api_host=api_host) + if _is_scoped_profile(profile): + return f"{key}-profile-{profile}" + return key + + +def _get_value_with_fallback(template, api_host, profile): + """Read the profile-scoped entry, with fallback to the legacy entry. + + Tokens stored before profile scoping existed live in unscoped + entries. The fallback keeps those sessions valid. Writes go to the + scoped entry, so the tokens migrate on the next refresh. + """ + value = _get_value(_format_key(template, api_host, profile)) + if value is None and _is_scoped_profile(profile): + value = _get_value(_format_key(template, api_host, None)) + return value + + def _sync_keyring_backend_env(): """Allow CLOUDSMITH_KEYRING_BACKEND to alias PYTHON_KEYRING_BACKEND.""" alias_value = os.environ.get("CLOUDSMITH_KEYRING_BACKEND") @@ -61,37 +86,72 @@ def _get_value(key): return None +def _import_macos_keychain(): + try: + from . import macos_keychain + except (OSError, AttributeError): + return None + return macos_keychain + + +def _effective_backend(): + backend = keyring.get_keyring() + chained_backends = getattr(backend, "backends", None) + if chained_backends: + return chained_backends[0] + return backend + + +def _update_keychain_item_in_place(service, username, value): + """Update the item with SecItemUpdate to keep its access control list. + + keyring deletes and re-creates the item on write. That resets the + access control list that the user approved in the keychain prompt. + """ + if sys.platform != "darwin": + return False + backend_module = type(_effective_backend()).__module__ + if not backend_module.startswith("keyring.backends.macOS"): + return False + keychain = _import_macos_keychain() + if keychain is None: + return False + return keychain.update_generic_password(service, username, value) + + def _set_value(key, value): _prepare_keyring_backend() username = _get_username() + if _update_keychain_item_in_place(key, username, value): + return keyring.set_password(key, username, value) -def store_access_token(api_host, access_token): - key = ACCESS_TOKEN_KEY.format(api_host=api_host) +def store_access_token(api_host, access_token, profile=None): + key = _format_key(ACCESS_TOKEN_KEY, api_host, profile) _set_value(key, access_token) -def get_access_token(api_host): +def get_access_token(api_host, profile=None): if not should_use_keyring(): return None - key = ACCESS_TOKEN_KEY.format(api_host=api_host) - return _get_value(key) + return _get_value_with_fallback(ACCESS_TOKEN_KEY, api_host, profile) -def update_refresh_attempted_at(api_host, refresh_time=None): +def update_refresh_attempted_at(api_host, refresh_time=None, profile=None): if refresh_time is None: refresh_time = datetime.now(tz=timezone.utc) refresh_attempted_at_value = refresh_time.isoformat() - key = ACCESS_TOKEN_REFRESH_ATTEMPTED_AT_KEY.format(api_host=api_host) + key = _format_key(ACCESS_TOKEN_REFRESH_ATTEMPTED_AT_KEY, api_host, profile) _set_value(key, refresh_attempted_at_value) -def get_refresh_attempted_at(api_host): - key = ACCESS_TOKEN_REFRESH_ATTEMPTED_AT_KEY.format(api_host=api_host) - value = _get_value(key) +def get_refresh_attempted_at(api_host, profile=None): + value = _get_value_with_fallback( + ACCESS_TOKEN_REFRESH_ATTEMPTED_AT_KEY, api_host, profile + ) if not value: return None @@ -102,11 +162,11 @@ def get_refresh_attempted_at(api_host): return None -def should_refresh_access_token(api_host): +def should_refresh_access_token(api_host, profile=None): if not should_use_keyring(): return False - token_refreshed_at = get_refresh_attempted_at(api_host) + token_refreshed_at = get_refresh_attempted_at(api_host, profile=profile) if token_refreshed_at: return token_refreshed_at < ( @@ -116,27 +176,30 @@ def should_refresh_access_token(api_host): return True -def store_refresh_token(api_host, refresh_token): - key = REFRESH_TOKEN_KEY.format(api_host=api_host) +def store_refresh_token(api_host, refresh_token, profile=None): + key = _format_key(REFRESH_TOKEN_KEY, api_host, profile) _set_value(key, refresh_token) -def get_refresh_token(api_host): - key = REFRESH_TOKEN_KEY.format(api_host=api_host) - return _get_value(key) +def get_refresh_token(api_host, profile=None): + return _get_value_with_fallback(REFRESH_TOKEN_KEY, api_host, profile) -def store_sso_tokens(api_host, access_token, refresh_token): +def store_sso_tokens(api_host, access_token, refresh_token, profile=None): """Store SSO tokens in keyring if enabled.""" if not should_use_keyring(): return False if access_token: - store_access_token(api_host=api_host, access_token=access_token) - update_refresh_attempted_at(api_host=api_host) + store_access_token( + api_host=api_host, access_token=access_token, profile=profile + ) + update_refresh_attempted_at(api_host=api_host, profile=profile) if refresh_token: - store_refresh_token(api_host=api_host, refresh_token=refresh_token) + store_refresh_token( + api_host=api_host, refresh_token=refresh_token, profile=profile + ) return True @@ -151,25 +214,39 @@ def _delete_value(key): return False -def _sso_keys(api_host): - """Return the keyring service names for all SSO-related entries.""" - return [ - ACCESS_TOKEN_KEY.format(api_host=api_host), - REFRESH_TOKEN_KEY.format(api_host=api_host), - ACCESS_TOKEN_REFRESH_ATTEMPTED_AT_KEY.format(api_host=api_host), +def _sso_keys(api_host, profile=None, include_legacy=True): + """Return the keyring service names for all SSO-related entries. + + For a scoped profile with include_legacy, the list also contains the + legacy unscoped entries so that the caller removes pre-scoping + sessions. + """ + templates = [ + ACCESS_TOKEN_KEY, + REFRESH_TOKEN_KEY, + ACCESS_TOKEN_REFRESH_ATTEMPTED_AT_KEY, ] + keys = [_format_key(template, api_host, profile) for template in templates] + if include_legacy and _is_scoped_profile(profile): + keys += [_format_key(template, api_host, None) for template in templates] + return keys -def has_sso_tokens(api_host): +def has_sso_tokens(api_host, profile=None): """Check if any SSO tokens exist in the keyring for the given host.""" if not should_use_keyring(): return False - return any(_get_value(key) for key in _sso_keys(api_host)) + return any(_get_value(key) for key in _sso_keys(api_host, profile=profile)) + +def delete_sso_tokens(api_host, profile=None, include_legacy=True): + """Delete all SSO tokens from the keyring for the given host. -def delete_sso_tokens(api_host): - """Delete all SSO tokens from the keyring for the given host.""" - results = [_delete_value(key) for key in _sso_keys(api_host)] + Set include_legacy to False to keep the legacy unscoped entries, + which hold the default profile's session. + """ + keys = _sso_keys(api_host, profile=profile, include_legacy=include_legacy) + results = [_delete_value(key) for key in keys] return any(results) diff --git a/cloudsmith_cli/core/macos_keychain.py b/cloudsmith_cli/core/macos_keychain.py new file mode 100644 index 00000000..8a4d22f4 --- /dev/null +++ b/cloudsmith_cli/core/macos_keychain.py @@ -0,0 +1,168 @@ +"""In-place updates for macOS keychain items. + +The keyring library implements each write as a delete followed by a +re-create (see set_generic_password in keyring/backends/macOS/api.py, +https://github.com/jaraco/keyring). macOS attaches the access control +list to the item, so the re-created item forgets every access grant +the user approved and the keychain prompts return on the next read. +SecItemUpdate changes the stored secret on the existing item and +keeps the access control list intact. + +The built-in backend cannot give this behavior: set_password is +hard-coded to delete and re-create, the backend exposes no update +function, and no configuration option changes the write path. +keyring 25.7.0, current at the time of this change, contains no fix. +The upstream reports https://github.com/jaraco/keyring/issues/619 +and https://github.com/jaraco/keyring/issues/512 describe the +resulting prompt storm but not the access control list reset that +causes it, so this module binds SecItemUpdate itself. + +This module does not register a keyring backend, and the backend +discovery is unchanged. cloudsmith_cli.core.keyring tries +update_generic_password before each write and falls back to the +normal keyring write when the item does not exist or the update +fails. + +The Security and CoreFoundation bindings load on the first call, not +at import time, so the module imports cleanly on every platform. + +The kSec* constants mirror the attributes that the keyring backend +stores, so an update targets exactly the items that keyring creates +and reads: + +- kSecClass with kSecClassGenericPassword selects the item class. +- kSecAttrService and kSecAttrAccount identify one item; keyring + stores its service name and username in them. +- kSecValueData holds the secret payload. + +See "Searching for keychain items": +https://developer.apple.com/documentation/security/keychain_services/keychain_items/searching_for_keychain_items +and SecItemUpdate: +https://developer.apple.com/documentation/security/1393617-secitemupdate +""" + +import ctypes +import functools +from ctypes import c_int32, c_void_p +from ctypes.util import find_library +from types import SimpleNamespace + +_KCF_STRING_ENCODING_UTF8 = 0x08000100 +_ERR_SEC_SUCCESS = 0 + + +@functools.cache +def _get_bindings(): + security = ctypes.CDLL(find_library("Security")) + core_foundation = ctypes.CDLL(find_library("CoreFoundation")) + + cf_string_create = core_foundation.CFStringCreateWithCString + cf_string_create.restype = c_void_p + cf_string_create.argtypes = (c_void_p, ctypes.c_char_p, ctypes.c_uint32) + + cf_data_create = core_foundation.CFDataCreate + cf_data_create.restype = c_void_p + cf_data_create.argtypes = (c_void_p, ctypes.c_char_p, ctypes.c_long) + + cf_dictionary_create = core_foundation.CFDictionaryCreate + cf_dictionary_create.restype = c_void_p + cf_dictionary_create.argtypes = ( + c_void_p, + c_void_p, + c_void_p, + ctypes.c_long, + c_void_p, + c_void_p, + ) + + cf_release = core_foundation.CFRelease + cf_release.restype = None + cf_release.argtypes = (c_void_p,) + + sec_item_update = security.SecItemUpdate + sec_item_update.restype = c_int32 + sec_item_update.argtypes = (c_void_p, c_void_p) + + return SimpleNamespace( + security=security, + cf_string_create=cf_string_create, + cf_data_create=cf_data_create, + cf_dictionary_create=cf_dictionary_create, + cf_release=cf_release, + sec_item_update=sec_item_update, + dictionary_key_callbacks=core_foundation.kCFTypeDictionaryKeyCallBacks, + dictionary_value_callbacks=core_foundation.kCFTypeDictionaryValueCallBacks, + ) + + +def _security_constant(bindings, name): + return c_void_p.in_dll(bindings.security, name) + + +def _cf_string(bindings, value): + return bindings.cf_string_create( + None, value.encode("utf-8"), _KCF_STRING_ENCODING_UTF8 + ) + + +def _cf_dictionary(bindings, pairs): + keys = (c_void_p * len(pairs))(*(key for key, _ in pairs)) + values = (c_void_p * len(pairs))(*(value for _, value in pairs)) + return bindings.cf_dictionary_create( + None, + keys, + values, + len(pairs), + bindings.dictionary_key_callbacks, + bindings.dictionary_value_callbacks, + ) + + +def _release(bindings, refs): + for ref in refs: + if ref: + bindings.cf_release(ref) + + +def update_generic_password(service, account, value): + """Update an existing generic password item in place. + + SecItemUpdate takes two dictionaries: a query that identifies the + item (class, service, account) and the attributes to change (the + secret data). Return True when the update succeeds. Return False + when the item does not exist, when macOS rejects the update, or + when the Security framework is not available. + """ + try: + bindings = _get_bindings() + class_key = _security_constant(bindings, "kSecClass") + class_value = _security_constant(bindings, "kSecClassGenericPassword") + service_key = _security_constant(bindings, "kSecAttrService") + account_key = _security_constant(bindings, "kSecAttrAccount") + value_key = _security_constant(bindings, "kSecValueData") + except (OSError, AttributeError, ValueError): + return False + encoded_value = value.encode("utf-8") + service_ref = _cf_string(bindings, service) + account_ref = _cf_string(bindings, account) + value_ref = bindings.cf_data_create(None, encoded_value, len(encoded_value)) + query = None + attributes = None + try: + if not (service_ref and account_ref and value_ref): + return False + query = _cf_dictionary( + bindings, + [ + (class_key, class_value), + (service_key, service_ref), + (account_key, account_ref), + ], + ) + attributes = _cf_dictionary(bindings, [(value_key, value_ref)]) + if not (query and attributes): + return False + status = bindings.sec_item_update(query, attributes) + return status == _ERR_SEC_SUCCESS + finally: + _release(bindings, (query, attributes, service_ref, account_ref, value_ref)) diff --git a/cloudsmith_cli/core/tests/test_keyring.py b/cloudsmith_cli/core/tests/test_keyring.py index 15bd7a4c..2a26c441 100644 --- a/cloudsmith_cli/core/tests/test_keyring.py +++ b/cloudsmith_cli/core/tests/test_keyring.py @@ -1,13 +1,15 @@ import getpass +import importlib import os from datetime import datetime, timedelta, timezone -from unittest.mock import ANY, patch +from unittest.mock import ANY, Mock, patch import keyring import pytest from freezegun import freeze_time from keyring.errors import KeyringError +from .. import keyring as core_keyring from ..keyring import ( delete_sso_tokens, get_access_token, @@ -235,6 +237,305 @@ def test_store_sso_tokens_returns_false_when_keyring_disabled( mock_set_password.assert_not_called() +class TestProfileScopedKeys: + """Tests for profile-scoped keyring service names.""" + + api_host = "https://example.com" + + def test_store_access_token_with_profile(self, mock_get_user, mock_set_password): + store_access_token(self.api_host, "access_token", profile="staging") + + mock_set_password.assert_called_once_with( + "cloudsmith_cli-access_token-https://example.com-profile-staging", + "test_user", + "access_token", + ) + + def test_store_access_token_with_default_profile_uses_legacy_key( + self, mock_get_user, mock_set_password + ): + store_access_token(self.api_host, "access_token", profile="default") + + mock_set_password.assert_called_once_with( + "cloudsmith_cli-access_token-https://example.com", + "test_user", + "access_token", + ) + + def test_get_access_token_with_profile(self, mock_get_user, mock_get_password): + mock_get_password.return_value = "access_token" + + assert get_access_token(self.api_host, profile="staging") == "access_token" + mock_get_password.assert_called_once_with( + "cloudsmith_cli-access_token-https://example.com-profile-staging", + "test_user", + ) + + def test_get_refresh_token_with_profile(self, mock_get_user, mock_get_password): + mock_get_password.return_value = "refresh_token" + + assert get_refresh_token(self.api_host, profile="staging") == "refresh_token" + mock_get_password.assert_called_once_with( + "cloudsmith_cli-refresh_token-https://example.com-profile-staging", + "test_user", + ) + + @freeze_time("2024-06-01 10:00:00") + def test_store_sso_tokens_with_profile(self, mock_get_user, mock_set_password): + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with patch.dict(os.environ, env, clear=True): + result = store_sso_tokens( + self.api_host, "access_token", "refresh_token", profile="staging" + ) + + assert result is True + assert mock_set_password.call_count == 3 + mock_set_password.assert_any_call( + "cloudsmith_cli-access_token-https://example.com-profile-staging", + "test_user", + "access_token", + ) + mock_set_password.assert_any_call( + "cloudsmith_cli-access_token_refresh_attempted_at-https://example.com" + "-profile-staging", + "test_user", + ANY, + ) + mock_set_password.assert_any_call( + "cloudsmith_cli-refresh_token-https://example.com-profile-staging", + "test_user", + "refresh_token", + ) + + def test_has_sso_tokens_with_profile(self, mock_get_user, mock_get_password): + mock_get_password.return_value = "some_token" + + assert has_sso_tokens(self.api_host, profile="staging") is True + first_key = mock_get_password.call_args_list[0].args[0] + assert first_key == ( + "cloudsmith_cli-access_token-https://example.com-profile-staging" + ) + + def test_delete_sso_tokens_with_profile_removes_legacy_entries( + self, mock_get_user, mock_delete_password + ): + assert delete_sso_tokens(self.api_host, profile="staging") is True + + deleted_keys = [call.args[0] for call in mock_delete_password.call_args_list] + assert deleted_keys == [ + "cloudsmith_cli-access_token-https://example.com-profile-staging", + "cloudsmith_cli-refresh_token-https://example.com-profile-staging", + "cloudsmith_cli-access_token_refresh_attempted_at-https://example.com" + "-profile-staging", + "cloudsmith_cli-access_token-https://example.com", + "cloudsmith_cli-refresh_token-https://example.com", + "cloudsmith_cli-access_token_refresh_attempted_at-https://example.com", + ] + + def test_delete_sso_tokens_without_legacy_keeps_unscoped_entries( + self, mock_get_user, mock_delete_password + ): + assert ( + delete_sso_tokens(self.api_host, profile="staging", include_legacy=False) + is True + ) + + deleted_keys = [call.args[0] for call in mock_delete_password.call_args_list] + assert deleted_keys == [ + "cloudsmith_cli-access_token-https://example.com-profile-staging", + "cloudsmith_cli-refresh_token-https://example.com-profile-staging", + "cloudsmith_cli-access_token_refresh_attempted_at-https://example.com" + "-profile-staging", + ] + + def test_get_access_token_with_profile_falls_back_to_legacy_key( + self, mock_get_user, mock_get_password + ): + mock_get_password.side_effect = [None, "legacy_token"] + + assert get_access_token(self.api_host, profile="staging") == "legacy_token" + requested_keys = [call.args[0] for call in mock_get_password.call_args_list] + assert requested_keys == [ + "cloudsmith_cli-access_token-https://example.com-profile-staging", + "cloudsmith_cli-access_token-https://example.com", + ] + + def test_get_access_token_with_default_profile_does_not_fall_back( + self, mock_get_user, mock_get_password + ): + mock_get_password.return_value = None + + assert get_access_token(self.api_host, profile="default") is None + mock_get_password.assert_called_once_with( + "cloudsmith_cli-access_token-https://example.com", "test_user" + ) + + def test_get_refresh_token_with_profile_falls_back_to_legacy_key( + self, mock_get_user, mock_get_password + ): + mock_get_password.side_effect = [None, "legacy_refresh"] + + assert get_refresh_token(self.api_host, profile="staging") == "legacy_refresh" + requested_keys = [call.args[0] for call in mock_get_password.call_args_list] + assert requested_keys == [ + "cloudsmith_cli-refresh_token-https://example.com-profile-staging", + "cloudsmith_cli-refresh_token-https://example.com", + ] + + @freeze_time("2024-06-01 10:00:00") + def test_should_refresh_access_token_with_profile( + self, mock_get_user, mock_get_password + ): + mock_get_password.return_value = ( + datetime.now(tz=timezone.utc) - timedelta(minutes=31) + ).isoformat() + + assert should_refresh_access_token(self.api_host, profile="staging") + mock_get_password.assert_called_once_with( + "cloudsmith_cli-access_token_refresh_attempted_at-https://example.com" + "-profile-staging", + "test_user", + ) + + +def test_macos_keychain_imports_on_every_platform(): + """The binary build imports every bundled module on Linux.""" + module = importlib.import_module("cloudsmith_cli.core.macos_keychain") + result = module.update_generic_password( + "cloudsmith_cli-import-check", "nobody", "value" + ) + assert result is False + + +class FakeMacosBackend: + pass + + +FakeMacosBackend.__module__ = "keyring.backends.macOS" + + +class FakeChainerBackend: + def __init__(self, backends): + self.backends = backends + + +class TestUpdateKeychainItemInPlace: + """Tests for the in-place update path that keeps the access control list.""" + + api_host = "https://example.com" + + def test_set_value_skips_set_password_when_update_succeeds( + self, mock_get_user, mock_set_password + ): + with patch.object( + core_keyring, "_update_keychain_item_in_place", return_value=True + ): + store_access_token(self.api_host, "access_token") + mock_set_password.assert_not_called() + + def test_set_value_falls_back_when_update_fails( + self, mock_get_user, mock_set_password + ): + with patch.object( + core_keyring, "_update_keychain_item_in_place", return_value=False + ): + store_access_token(self.api_host, "access_token") + mock_set_password.assert_called_once_with( + "cloudsmith_cli-access_token-https://example.com", + "test_user", + "access_token", + ) + + def test_update_returns_false_off_macos(self, mock_get_keyring): + importer = Mock() + with ( + patch("sys.platform", "linux"), + patch.object(core_keyring, "_import_macos_keychain", importer), + ): + result = core_keyring._update_keychain_item_in_place( + "service", "user", "value" + ) + assert result is False + importer.assert_not_called() + + def test_update_returns_false_for_non_macos_backend(self, mock_get_keyring): + importer = Mock() + with ( + patch("sys.platform", "darwin"), + patch.object(core_keyring, "_import_macos_keychain", importer), + ): + result = core_keyring._update_keychain_item_in_place( + "service", "user", "value" + ) + assert result is False + importer.assert_not_called() + + def test_update_uses_macos_module_for_macos_backend(self, mock_get_keyring): + mock_get_keyring.return_value = FakeMacosBackend() + macos_module = Mock() + macos_module.update_generic_password.return_value = True + with ( + patch("sys.platform", "darwin"), + patch.object( + core_keyring, "_import_macos_keychain", return_value=macos_module + ), + ): + result = core_keyring._update_keychain_item_in_place( + "service", "user", "value" + ) + assert result is True + macos_module.update_generic_password.assert_called_once_with( + "service", "user", "value" + ) + + def test_update_uses_macos_module_when_chainer_delegates_to_macos( + self, mock_get_keyring + ): + mock_get_keyring.return_value = FakeChainerBackend([FakeMacosBackend(), Mock()]) + macos_module = Mock() + macos_module.update_generic_password.return_value = True + with ( + patch("sys.platform", "darwin"), + patch.object( + core_keyring, "_import_macos_keychain", return_value=macos_module + ), + ): + result = core_keyring._update_keychain_item_in_place( + "service", "user", "value" + ) + assert result is True + macos_module.update_generic_password.assert_called_once_with( + "service", "user", "value" + ) + + def test_update_returns_false_when_chainer_prefers_other_backend( + self, mock_get_keyring + ): + mock_get_keyring.return_value = FakeChainerBackend([Mock(), FakeMacosBackend()]) + importer = Mock() + with ( + patch("sys.platform", "darwin"), + patch.object(core_keyring, "_import_macos_keychain", importer), + ): + result = core_keyring._update_keychain_item_in_place( + "service", "user", "value" + ) + assert result is False + importer.assert_not_called() + + def test_update_returns_false_when_module_unavailable(self, mock_get_keyring): + mock_get_keyring.return_value = FakeMacosBackend() + with ( + patch("sys.platform", "darwin"), + patch.object(core_keyring, "_import_macos_keychain", return_value=None), + ): + result = core_keyring._update_keychain_item_in_place( + "service", "user", "value" + ) + assert result is False + + class TestShouldUseKeyring: """Tests for the should_use_keyring function.""" diff --git a/cloudsmith_cli/core/tests/test_keyring_provider.py b/cloudsmith_cli/core/tests/test_keyring_provider.py index 334a389b..546cf5a7 100644 --- a/cloudsmith_cli/core/tests/test_keyring_provider.py +++ b/cloudsmith_cli/core/tests/test_keyring_provider.py @@ -1,10 +1,15 @@ """Tests for the keyring credential provider.""" import os -from unittest.mock import MagicMock, patch +from unittest.mock import MagicMock, call, patch +import pytest + +from cloudsmith_cli.cli import saml +from cloudsmith_cli.core import keyring +from cloudsmith_cli.core.api.exceptions import ApiException from cloudsmith_cli.core.credentials.models import CredentialContext -from cloudsmith_cli.core.credentials.providers import KeyringProvider +from cloudsmith_cli.core.credentials.providers import KeyringProvider, keyring_provider class TestKeyringProvider: @@ -15,8 +20,6 @@ def test_returns_none_when_keyring_disabled(self): assert result is None def test_returns_none_when_no_token(self): - from cloudsmith_cli.core import keyring - provider = KeyringProvider() env = os.environ.copy() env.pop("CLOUDSMITH_NO_KEYRING", None) @@ -29,8 +32,6 @@ def test_returns_none_when_no_token(self): assert result is None def test_returns_bearer_token(self): - from cloudsmith_cli.core import keyring - provider = KeyringProvider() env = os.environ.copy() env.pop("CLOUDSMITH_NO_KEYRING", None) @@ -47,10 +48,6 @@ def test_returns_bearer_token(self): assert result.source_name == "keyring" def test_returns_none_on_refresh_failure(self): - from cloudsmith_cli.cli import saml - from cloudsmith_cli.core import keyring - from cloudsmith_cli.core.api.exceptions import ApiException - provider = KeyringProvider() context = CredentialContext(session=MagicMock()) env = os.environ.copy() @@ -71,3 +68,197 @@ def test_returns_none_on_refresh_failure(self): result = provider.resolve(context) assert result is None assert context.keyring_refresh_failed is True + + def test_passes_profile_to_keyring(self): + provider = KeyringProvider() + context = CredentialContext(session=MagicMock(), profile="staging") + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with ( + patch.dict(os.environ, env, clear=True), + patch.object(keyring, "should_use_keyring", return_value=True), + patch.object( + keyring, "get_access_token", return_value="old_token" + ) as get_access_mock, + patch.object( + keyring, "should_refresh_access_token", return_value=True + ) as should_refresh_mock, + patch.object( + keyring, "get_refresh_token", return_value="old_refresh" + ) as get_refresh_mock, + patch.object( + keyring_provider, + "refresh_access_token", + return_value=("new_token", "new_refresh"), + ), + patch.object(keyring, "store_sso_tokens") as store_mock, + ): + result = provider.resolve(context) + + assert result is not None + assert result.api_key == "new_token" + get_access_mock.assert_called_once_with(context.api_host, profile="staging") + should_refresh_mock.assert_called_once_with(context.api_host, profile="staging") + get_refresh_mock.assert_called_once_with(context.api_host, profile="staging") + store_mock.assert_called_once_with( + context.api_host, "new_token", "new_refresh", profile="staging" + ) + + @pytest.mark.parametrize("status", [400, 401, 403, 422]) + def test_wipes_tokens_when_refresh_is_rejected(self, status): + provider = KeyringProvider() + context = CredentialContext(session=MagicMock(), profile="staging") + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with ( + patch.dict(os.environ, env, clear=True), + patch.object(keyring, "should_use_keyring", return_value=True), + patch.object(keyring, "get_access_token", return_value="stale_token"), + patch.object(keyring, "should_refresh_access_token", return_value=True), + patch.object(keyring, "get_refresh_token", return_value="stale_refresh"), + patch.object( + keyring_provider, + "refresh_access_token", + side_effect=ApiException(status=status, detail="Rejected"), + ), + patch.object(keyring, "delete_sso_tokens") as delete_mock, + patch.object(keyring, "update_refresh_attempted_at") as attempted_mock, + ): + result = provider.resolve(context) + + assert result is None + assert context.keyring_refresh_failed is True + delete_mock.assert_called_once_with( + context.api_host, profile="staging", include_legacy=False + ) + attempted_mock.assert_not_called() + + def test_wipes_legacy_tokens_when_profile_has_none(self): + provider = KeyringProvider() + context = CredentialContext(session=MagicMock(), profile="staging") + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with ( + patch.dict(os.environ, env, clear=True), + patch.object(keyring, "should_use_keyring", return_value=True), + patch.object(keyring, "get_access_token", return_value="stale_token"), + patch.object(keyring, "should_refresh_access_token", return_value=True), + patch.object(keyring, "get_refresh_token", return_value="stale_refresh"), + patch.object( + keyring_provider, + "refresh_access_token", + side_effect=ApiException(status=401, detail="Rejected"), + ), + patch.object( + keyring, "delete_sso_tokens", side_effect=[False, True] + ) as delete_mock, + patch.object(keyring, "update_refresh_attempted_at") as attempted_mock, + ): + result = provider.resolve(context) + + assert result is None + assert context.keyring_refresh_failed is True + assert delete_mock.call_args_list == [ + call(context.api_host, profile="staging", include_legacy=False), + call(context.api_host), + ] + attempted_mock.assert_not_called() + + def test_stamps_attempt_when_wipe_removes_nothing(self): + provider = KeyringProvider() + context = CredentialContext(session=MagicMock(), profile="staging") + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with ( + patch.dict(os.environ, env, clear=True), + patch.object(keyring, "should_use_keyring", return_value=True), + patch.object(keyring, "get_access_token", return_value="stale_token"), + patch.object(keyring, "should_refresh_access_token", return_value=True), + patch.object(keyring, "get_refresh_token", return_value="stale_refresh"), + patch.object( + keyring_provider, + "refresh_access_token", + side_effect=ApiException(status=401, detail="Rejected"), + ), + patch.object(keyring, "delete_sso_tokens", return_value=False), + patch.object(keyring, "update_refresh_attempted_at") as attempted_mock, + ): + result = provider.resolve(context) + + assert result is None + assert context.keyring_refresh_failed is True + attempted_mock.assert_called_once_with(context.api_host, profile="staging") + + def test_skips_refresh_when_no_refresh_token_is_stored(self): + provider = KeyringProvider() + context = CredentialContext(session=MagicMock(), profile="staging") + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with ( + patch.dict(os.environ, env, clear=True), + patch.object(keyring, "should_use_keyring", return_value=True), + patch.object(keyring, "get_access_token", return_value="sso_token"), + patch.object(keyring, "should_refresh_access_token", return_value=True), + patch.object(keyring, "get_refresh_token", return_value=None), + patch.object(keyring_provider, "refresh_access_token") as refresh_mock, + patch.object(keyring, "delete_sso_tokens") as delete_mock, + ): + result = provider.resolve(context) + + assert result is not None + assert result.api_key == "sso_token" + assert context.keyring_refresh_failed is False + refresh_mock.assert_not_called() + delete_mock.assert_not_called() + + def test_keeps_tokens_on_transient_refresh_error(self): + provider = KeyringProvider() + context = CredentialContext(session=MagicMock(), profile="staging") + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with ( + patch.dict(os.environ, env, clear=True), + patch.object(keyring, "should_use_keyring", return_value=True), + patch.object(keyring, "get_access_token", return_value="stale_token"), + patch.object(keyring, "should_refresh_access_token", return_value=True), + patch.object(keyring, "get_refresh_token", return_value="stale_refresh"), + patch.object( + keyring_provider, + "refresh_access_token", + side_effect=ApiException(status=503, detail="Service Unavailable"), + ), + patch.object(keyring, "delete_sso_tokens") as delete_mock, + patch.object(keyring, "update_refresh_attempted_at") as attempted_mock, + ): + result = provider.resolve(context) + + assert result is None + assert context.keyring_refresh_failed is True + delete_mock.assert_not_called() + attempted_mock.assert_called_once_with(context.api_host, profile="staging") + + def test_refresh_without_access_token_in_response_is_a_failure(self): + provider = KeyringProvider() + context = CredentialContext(session=MagicMock(), profile="staging") + env = os.environ.copy() + env.pop("CLOUDSMITH_NO_KEYRING", None) + with ( + patch.dict(os.environ, env, clear=True), + patch.object(keyring, "should_use_keyring", return_value=True), + patch.object(keyring, "get_access_token", return_value="stale_token"), + patch.object(keyring, "should_refresh_access_token", return_value=True), + patch.object(keyring, "get_refresh_token", return_value="stale_refresh"), + patch.object( + keyring_provider, + "refresh_access_token", + return_value=(None, None), + ), + patch.object(keyring, "store_sso_tokens") as store_mock, + patch.object(keyring, "update_refresh_attempted_at") as attempted_mock, + ): + result = provider.resolve(context) + + assert result is None + assert context.keyring_refresh_failed is True + store_mock.assert_not_called() + attempted_mock.assert_called_once_with(context.api_host, profile="staging") diff --git a/cloudsmith_cli/core/tests/test_metadata.py b/cloudsmith_cli/core/tests/test_metadata.py index b2f1e27d..06632d7e 100644 --- a/cloudsmith_cli/core/tests/test_metadata.py +++ b/cloudsmith_cli/core/tests/test_metadata.py @@ -30,9 +30,11 @@ def _setup_api(monkeypatch): off the same Configuration singleton. Keyring is stubbed so we never touch the user's real SSO tokens during a test run. """ - monkeypatch.setattr(keyring, "get_access_token", lambda host: None) - monkeypatch.setattr(keyring, "get_refresh_token", lambda host: None) - monkeypatch.setattr(keyring, "should_refresh_access_token", lambda host: False) + monkeypatch.setattr(keyring, "get_access_token", lambda host, profile=None: None) + monkeypatch.setattr(keyring, "get_refresh_token", lambda host, profile=None: None) + monkeypatch.setattr( + keyring, "should_refresh_access_token", lambda host, profile=None: False + ) monkeypatch.setattr( httpretty.core.fakesock.socket, "shutdown", From e681bd31060f2441f014f82f6b311b119538632a Mon Sep 17 00:00:00 2001 From: "copilot-swe-agent[bot]" <198982749+Copilot@users.noreply.github.com> Date: Tue, 25 Aug 2026 15:41:36 +0000 Subject: [PATCH 2/7] fix: restore refresh token import in keyring provider Co-authored-by: cloudsmith-iduffy <178375997+cloudsmith-iduffy@users.noreply.github.com> --- cloudsmith_cli/core/credentials/providers/keyring_provider.py | 1 + 1 file changed, 1 insertion(+) diff --git a/cloudsmith_cli/core/credentials/providers/keyring_provider.py b/cloudsmith_cli/core/credentials/providers/keyring_provider.py index c8004789..1d76de21 100644 --- a/cloudsmith_cli/core/credentials/providers/keyring_provider.py +++ b/cloudsmith_cli/core/credentials/providers/keyring_provider.py @@ -4,6 +4,7 @@ import logging +from ....cli.saml import refresh_access_token from ....core import keyring from ...api.exceptions import ApiException from ..models import CredentialContext, CredentialResult From 7797700cce3999d1bbf78407818c1db1aea2de39 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Tue, 25 Aug 2026 18:00:51 +0100 Subject: [PATCH 3/7] fix: defer credential provider imports and fix logout test assertion The credential provider chain imported all providers at module level, which pulled requests and cloudsmith_api into every CLI invocation through the keyring provider's SAML dependency. Defer that import to CredentialProviderChain.__init__ so it only loads when the chain is built. Also fix a logout test assertion that dropped the profile=None keyword argument the command actually passes. --- cloudsmith_cli/cli/tests/commands/test_logout.py | 2 +- cloudsmith_cli/core/credentials/chain.py | 16 ++++++++-------- 2 files changed, 9 insertions(+), 9 deletions(-) diff --git a/cloudsmith_cli/cli/tests/commands/test_logout.py b/cloudsmith_cli/cli/tests/commands/test_logout.py index 4e03a568..8b58a52e 100644 --- a/cloudsmith_cli/cli/tests/commands/test_logout.py +++ b/cloudsmith_cli/cli/tests/commands/test_logout.py @@ -60,7 +60,7 @@ 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) def test_dry_run(self, runner, mock_deps): mock_creds, mock_keyring = mock_deps diff --git a/cloudsmith_cli/core/credentials/chain.py b/cloudsmith_cli/core/credentials/chain.py index 3b5e3b09..e4ae58ae 100644 --- a/cloudsmith_cli/core/credentials/chain.py +++ b/cloudsmith_cli/core/credentials/chain.py @@ -9,14 +9,6 @@ import logging from typing import TYPE_CHECKING -from .providers import ( - CLIFlagProvider, - CredentialsFileProvider, - EnvVarProvider, - KeyringProvider, - OidcProvider, -) - if TYPE_CHECKING: from .models import CredentialContext, CredentialResult from .provider import CredentialProvider @@ -35,6 +27,14 @@ def __init__(self, providers: list[CredentialProvider] | None = None): if providers is not None: self.providers = providers else: + from .providers import ( + CLIFlagProvider, + CredentialsFileProvider, + EnvVarProvider, + KeyringProvider, + OidcProvider, + ) + self.providers = [ CLIFlagProvider(), EnvVarProvider(), From 9e717bbf6e16f59850feb99628fc33dbee306ee9 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Wed, 26 Aug 2026 14:47:48 +0100 Subject: [PATCH 4/7] fix: keep the default profile's session on scoped-profile logout Logout deleted the legacy unscoped entries together with the profile's scoped entries. The unscoped entries hold the default profile's session, so a scoped-profile logout ended both sessions. Delete the legacy entries only when the profile has no scoped entries, as the refresh-rejection path already does. Co-Authored-By: Claude Fable 5 --- cloudsmith_cli/cli/commands/logout.py | 11 ++++-- .../cli/tests/commands/test_logout.py | 36 +++++++++++++++++-- 2 files changed, 42 insertions(+), 5 deletions(-) diff --git a/cloudsmith_cli/cli/commands/logout.py b/cloudsmith_cli/cli/commands/logout.py index 76914a5b..ba4a784a 100644 --- a/cloudsmith_cli/cli/commands/logout.py +++ b/cloudsmith_cli/cli/commands/logout.py @@ -34,7 +34,12 @@ def _clear_credentials(dry_run, use_stderr): def _clear_keyring(api_host, dry_run, use_stderr, profile=None): - """Clear SSO tokens from keyring. Returns result dict.""" + """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).", @@ -51,7 +56,9 @@ def _clear_keyring(api_host, dry_run, use_stderr, profile=None): click.echo("Would remove SSO tokens from system keyring.", err=use_stderr) return {"action": "would_remove"} - deleted = keyring.delete_sso_tokens(api_host, profile=profile) + 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) diff --git a/cloudsmith_cli/cli/tests/commands/test_logout.py b/cloudsmith_cli/cli/tests/commands/test_logout.py index 8b58a52e..74b3c0c2 100644 --- a/cloudsmith_cli/cli/tests/commands/test_logout.py +++ b/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 @@ -50,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, profile=None) + 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 @@ -60,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, profile=None) + 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 From e502f93cef8201cb58acd66cd3b59c3a731ac818 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Wed, 26 Aug 2026 14:47:57 +0100 Subject: [PATCH 5/7] refactor: move the SSO refresh helpers into core The keyring provider imported refresh_access_token from cli.saml, so core depended on cli at import time. Move refresh_access_token and raise_for_api_error into core/sso.py. cli/saml.py re-exports them for its existing callers. The deferred provider import in chain.py stays. decorators.py imports chain on every CLI start, so a module-level provider import would load requests and cloudsmith_api at startup and fail test_cli_import_does_not_load_heavy_modules. Co-Authored-By: Claude Fable 5 --- cloudsmith_cli/cli/saml.py | 54 ++++--------------- .../credentials/providers/keyring_provider.py | 2 +- cloudsmith_cli/core/sso.py | 49 +++++++++++++++++ pyproject.toml | 1 + 4 files changed, 60 insertions(+), 46 deletions(-) create mode 100644 cloudsmith_cli/core/sso.py diff --git a/cloudsmith_cli/cli/saml.py b/cloudsmith_cli/cli/saml.py index 28bca072..053bbe36 100644 --- a/cloudsmith_cli/cli/saml.py +++ b/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) diff --git a/cloudsmith_cli/core/credentials/providers/keyring_provider.py b/cloudsmith_cli/core/credentials/providers/keyring_provider.py index 1d76de21..a4c2f7a4 100644 --- a/cloudsmith_cli/core/credentials/providers/keyring_provider.py +++ b/cloudsmith_cli/core/credentials/providers/keyring_provider.py @@ -4,9 +4,9 @@ import logging -from ....cli.saml import refresh_access_token from ....core import keyring from ...api.exceptions import ApiException +from ...sso import refresh_access_token from ..models import CredentialContext, CredentialResult from ..provider import CredentialProvider diff --git a/cloudsmith_cli/core/sso.py b/cloudsmith_cli/core/sso.py new file mode 100644 index 00000000..a8b4c233 --- /dev/null +++ b/cloudsmith_cli/core/sso.py @@ -0,0 +1,49 @@ +"""SSO token refresh against the Cloudsmith API.""" + +import requests + +from .api.exceptions import ApiException + + +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 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) diff --git a/pyproject.toml b/pyproject.toml index f65d1bcf..c8c8c36e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -178,3 +178,4 @@ banned-module-level-imports = [ 'cloudsmith_cli/core/mcp/*' = ["TID253"] 'cloudsmith_cli/core/rest.py' = ["TID253"] 'cloudsmith_cli/core/session.py' = ["TID253"] +'cloudsmith_cli/core/sso.py' = ["TID253"] From 47e1c7228a12d959ec3c44275422fe6603edcf76 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Wed, 26 Aug 2026 14:48:05 +0100 Subject: [PATCH 6/7] test: patch the provider's refresh binding, mock the keyring backend test_returns_none_on_refresh_failure patched saml.refresh_access_token, but the provider binds the function at module level, so the patch did not intercept and the 401 path never ran. On macOS the test could reach the real keychain. Patch keyring_provider.refresh_access_token, patch delete_sso_tokens, and mock get_keyring for the whole file. Co-Authored-By: Claude Fable 5 --- cloudsmith_cli/core/tests/test_keyring_provider.py | 12 ++++++++++-- 1 file changed, 10 insertions(+), 2 deletions(-) diff --git a/cloudsmith_cli/core/tests/test_keyring_provider.py b/cloudsmith_cli/core/tests/test_keyring_provider.py index 546cf5a7..6c1ec47d 100644 --- a/cloudsmith_cli/core/tests/test_keyring_provider.py +++ b/cloudsmith_cli/core/tests/test_keyring_provider.py @@ -5,13 +5,20 @@ import pytest -from cloudsmith_cli.cli import saml from cloudsmith_cli.core import keyring from cloudsmith_cli.core.api.exceptions import ApiException from cloudsmith_cli.core.credentials.models import CredentialContext from cloudsmith_cli.core.credentials.providers import KeyringProvider, keyring_provider +@pytest.fixture(autouse=True) +def mock_get_keyring(): + import keyring as keyring_backend + + with patch.object(keyring_backend, "get_keyring") as get_keyring_mock: + yield get_keyring_mock + + class TestKeyringProvider: def test_returns_none_when_keyring_disabled(self): provider = KeyringProvider() @@ -59,10 +66,11 @@ def test_returns_none_on_refresh_failure(self): patch.object(keyring, "should_refresh_access_token", return_value=True), patch.object(keyring, "get_refresh_token", return_value="refresh_tok"), patch.object( - saml, + keyring_provider, "refresh_access_token", side_effect=ApiException(status=401, detail="Unauthorized"), ), + patch.object(keyring, "delete_sso_tokens", return_value=True), patch.object(keyring, "update_refresh_attempted_at"), ): result = provider.resolve(context) From 06313f2d680f0026979fdc37706c001a97412b40 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Wed, 26 Aug 2026 16:54:14 +0100 Subject: [PATCH 7/7] fix: include auth scheme in Cargo credentials (#394) Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com> --- .../cli/tests/test_credential_helper_cargo.py | 26 ++++++--- .../credentials/providers/oidc_provider.py | 2 + .../core/tests/test_oidc_provider.py | 54 +++++++++++++++++++ .../credential_helpers/cargo/runtime.py | 18 +++++-- 4 files changed, 91 insertions(+), 9 deletions(-) create mode 100644 cloudsmith_cli/core/tests/test_oidc_provider.py diff --git a/cloudsmith_cli/cli/tests/test_credential_helper_cargo.py b/cloudsmith_cli/cli/tests/test_credential_helper_cargo.py index 5ea6a4ae..5987ddd9 100644 --- a/cloudsmith_cli/cli/tests/test_credential_helper_cargo.py +++ b/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): diff --git a/cloudsmith_cli/core/credentials/providers/oidc_provider.py b/cloudsmith_cli/core/credentials/providers/oidc_provider.py index d2eee69f..680571b6 100644 --- a/cloudsmith_cli/core/credentials/providers/oidc_provider.py +++ b/cloudsmith_cli/core/credentials/providers/oidc_provider.py @@ -54,6 +54,7 @@ def resolve( # pylint: disable=too-many-return-statements api_key=cached, source_name="oidc", source_detail=f"OIDC [cached] (org: {org}, service: {service_slug})", + auth_type="bearer", ) from ..oidc.detectors import detect_environment @@ -112,4 +113,5 @@ def resolve( # pylint: disable=too-many-return-statements api_key=cloudsmith_token, source_name="oidc", source_detail=f"OIDC via {detector.name} (org: {org}, service: {service_slug})", + auth_type="bearer", ) diff --git a/cloudsmith_cli/core/tests/test_oidc_provider.py b/cloudsmith_cli/core/tests/test_oidc_provider.py new file mode 100644 index 00000000..734fdddc --- /dev/null +++ b/cloudsmith_cli/core/tests/test_oidc_provider.py @@ -0,0 +1,54 @@ +"""Tests for the OIDC credential provider.""" + +from unittest.mock import Mock, patch + +from cloudsmith_cli.core.credentials.models import CredentialContext +from cloudsmith_cli.core.credentials.providers.oidc_provider import OidcProvider + + +def _context() -> CredentialContext: + return CredentialContext( + org="cloudsmith", + oidc_service_slug="github-actions", + ) + + +def test_cached_oidc_token_is_a_bearer_credential(): + """Cached exchanged tokens retain their HTTP authorization scheme.""" + with patch( + "cloudsmith_cli.core.credentials.oidc.cache.get_cached_token", + return_value="cached-token", + ): + credential = OidcProvider().resolve(_context()) + + assert credential is not None + assert credential.auth_type == "bearer" + + +def test_exchanged_oidc_token_is_a_bearer_credential(): + """Freshly exchanged tokens use Bearer authentication.""" + detector = Mock(name="github-actions") + detector.name = "github-actions" + detector.get_token.return_value = "vendor-token" + + with ( + patch( + "cloudsmith_cli.core.credentials.oidc.cache.get_cached_token", + return_value=None, + ), + patch( + "cloudsmith_cli.core.credentials.oidc.cache.store_cached_token", + ), + patch( + "cloudsmith_cli.core.credentials.oidc.detectors.detect_environment", + return_value=detector, + ), + patch( + "cloudsmith_cli.core.credentials.oidc.exchange.exchange_oidc_token", + return_value="exchanged-token", + ), + ): + credential = OidcProvider().resolve(_context()) + + assert credential is not None + assert credential.auth_type == "bearer" diff --git a/cloudsmith_cli/credential_helpers/cargo/runtime.py b/cloudsmith_cli/credential_helpers/cargo/runtime.py index 97e742bd..d6587543 100644 --- a/cloudsmith_cli/credential_helpers/cargo/runtime.py +++ b/cloudsmith_cli/credential_helpers/cargo/runtime.py @@ -99,7 +99,8 @@ def get_credentials(index_url, credential=None, api_host=None, org=None): org: Organisation slug whose custom domains to match against Returns: - str: The token Cargo sends as its ``Authorization`` header, or None + str: The complete value Cargo sends as its ``Authorization`` header, + or None """ if not credential or not credential.api_key: return None @@ -109,7 +110,18 @@ def get_credentials(index_url, credential=None, api_host=None, org=None): ): return None - return credential.api_key + return _authorization_value(credential) + + +def _authorization_value(credential) -> str: + """Return the complete HTTP Authorization value for *credential*. + + Cargo forwards the credential provider's ``token`` field verbatim as the + Authorization header. Cloudsmith API credentials use the ``token`` scheme, + while SSO credentials use ``Bearer``. + """ + scheme = "Bearer" if credential.auth_type == "bearer" else "token" + return f"{scheme} {credential.api_key}" def _handle_get(request, credential, api_host, org) -> dict: @@ -137,7 +149,7 @@ def _handle_get(request, credential, api_host, org) -> dict: return _ok( { "kind": "get", - "token": credential.api_key, + "token": _authorization_value(credential), # A Cloudsmith token is organisation-wide and not scoped to the # read/publish/yank/owners operation, so it is cacheable for the # session and across operations.