From 7e3c0e33f4f15021418b6684687b7ce1b8c6e397 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Sun, 23 Aug 2026 12:38:57 +0100 Subject: [PATCH 1/3] 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/3] 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/3] 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(),