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 9ac2199d..76914a5b 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) @@ -131,7 +131,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 7062248e..8b58a52e 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 @@ -59,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/cli/tests/test_webserver.py b/cloudsmith_cli/cli/tests/test_webserver.py index a194195e..cae63cea 100644 --- a/cloudsmith_cli/cli/tests/test_webserver.py +++ b/cloudsmith_cli/cli/tests/test_webserver.py @@ -64,6 +64,7 @@ def mock_handler(self): ) handler.server_instance = MagicMock() handler.server_instance.api_host = "https://api.cloudsmith.io" + handler.server_instance.profile = None handler.refresh_api_on_success = False handler.session = MagicMock() handler.debug = False @@ -99,6 +100,41 @@ def test_store_sso_tokens_called_when_keyring_enabled(self, mock_handler): "https://api.cloudsmith.io", "test_access_token", "test_refresh_token", + profile=None, + ) + + def test_store_sso_tokens_receives_profile(self, mock_handler): + """Verify store_sso_tokens receives the profile from the server.""" + mock_handler.server_instance.profile = "staging" + with ( + patch( + "cloudsmith_cli.cli.webserver.store_sso_tokens", return_value=True + ) as mock_store, + patch.object(mock_handler, "_return_success_response"), + patch.object( + AuthenticationWebRequestHandler, + "query_data", + new_callable=PropertyMock, + ) as mock_query, + patch.object( + AuthenticationWebRequestHandler, + "api_host", + new_callable=PropertyMock, + ) as mock_host, + ): + mock_query.return_value = { + "access_token": "test_access_token", + "refresh_token": "test_refresh_token", + } + mock_host.return_value = "https://api.cloudsmith.io" + + mock_handler.do_GET() + + mock_store.assert_called_once_with( + "https://api.cloudsmith.io", + "test_access_token", + "test_refresh_token", + profile="staging", ) def test_message_shown_when_keyring_disabled(self, mock_handler): diff --git a/cloudsmith_cli/cli/webserver.py b/cloudsmith_cli/cli/webserver.py index 242c967c..c01f4ef4 100644 --- a/cloudsmith_cli/cli/webserver.py +++ b/cloudsmith_cli/cli/webserver.py @@ -35,6 +35,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 @@ -138,6 +139,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 _prompt_and_exchange_2fa_token(self, two_factor_token): """Prompt for a 2FA code, and prompt again while the API rejects it.""" click.echo(err=True) @@ -229,6 +235,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") @@ -244,21 +268,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 @@ -272,19 +282,7 @@ def do_GET(self): two_factor_token ) - # 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) return except Exception: 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(), diff --git a/cloudsmith_cli/core/credentials/providers/keyring_provider.py b/cloudsmith_cli/core/credentials/providers/keyring_provider.py index 143e631b..1d76de21 100644 --- a/cloudsmith_cli/core/credentials/providers/keyring_provider.py +++ b/cloudsmith_cli/core/credentials/providers/keyring_provider.py @@ -4,12 +4,38 @@ import logging +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.""" @@ -21,35 +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: - from ....cli.saml import refresh_access_token - - 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 8b186472..6919a013 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 ACCESS_TOKEN_KEY = "cloudsmith_cli-access_token-{api_host}" @@ -21,6 +22,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") @@ -100,39 +125,76 @@ 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(): + import keyring + + 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): import keyring _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 @@ -143,11 +205,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 < ( @@ -157,27 +219,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 @@ -195,25 +260,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): - """Delete all SSO tokens from the keyring for the given host.""" - results = [_delete_value(key) for key in _sso_keys(api_host)] +def delete_sso_tokens(api_host, profile=None, include_legacy=True): + """Delete all SSO tokens from the keyring for the given 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 485fe977..ee5d1f09 100644 --- a/cloudsmith_cli/core/tests/test_keyring.py +++ b/cloudsmith_cli/core/tests/test_keyring.py @@ -1,12 +1,14 @@ 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 pytest from freezegun import freeze_time from keyrings.cryptfile.cryptfile import CryptFileKeyring +from .. import keyring as core_keyring from ..keyring import ( delete_sso_tokens, get_access_token, @@ -248,6 +250,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 01778a85..92e07c3a 100644 --- a/cloudsmith_cli/core/tests/test_metadata.py +++ b/cloudsmith_cli/core/tests/test_metadata.py @@ -29,9 +29,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",