From 6a121de95ecc9801f60fbf9e73652e3ac344b782 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Thu, 3 Sep 2026 00:18:14 +0100 Subject: [PATCH 1/3] Harden automatic SSO renewal --- CHANGELOG.md | 4 + cloudsmith_cli/cli/decorators.py | 36 ++- cloudsmith_cli/cli/tests/test_decorators.py | 68 ++++- cloudsmith_cli/core/credentials/models.py | 3 + .../credentials/providers/keyring_provider.py | 128 ++++----- cloudsmith_cli/core/keyring.py | 38 ++- cloudsmith_cli/core/sso.py | 172 ++++++++++++ cloudsmith_cli/core/tests/test_keyring.py | 56 ++-- .../core/tests/test_keyring_provider.py | 255 ++---------------- cloudsmith_cli/core/tests/test_sso.py | 117 ++++++++ 10 files changed, 542 insertions(+), 335 deletions(-) create mode 100644 cloudsmith_cli/core/tests/test_sso.py diff --git a/CHANGELOG.md b/CHANGELOG.md index a4b10ebc..790c6efe 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -15,6 +15,10 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0. - `cloudsmith domains list` now includes the Workspace slug in a `workspace` field. +### Fixed + +- SSO access tokens now refresh based on their JWT expiry. Transient refresh failures retain usable tokens, while expired or rejected sessions fall back to other authentication. + ## [1.26.0] - 2026-08-26 ### Added diff --git a/cloudsmith_cli/cli/decorators.py b/cloudsmith_cli/cli/decorators.py index a76305e8..f5124cfb 100644 --- a/cloudsmith_cli/cli/decorators.py +++ b/cloudsmith_cli/cli/decorators.py @@ -495,6 +495,7 @@ def wrapper(ctx, *args, **kwargs): opts.oidc_detector_order, oidc_disabled_detectors ) + is_auth_command = ctx.command.name in ("authenticate", "login") context = CredentialContext( session=opts.session, api_key_from_flag=opts.api_key_from_flag, @@ -510,24 +511,37 @@ def wrapper(ctx, *args, **kwargs): oidc_discovery_disabled=opts.oidc_discovery_disabled, oidc_detector_order=opts.oidc_detector_order, oidc_disabled_detectors=oidc_disabled_detectors, + skip_keyring_refresh=is_auth_command, ) chain = CredentialProviderChain() credential = chain.resolve(context) - if context.keyring_refresh_failed: - click.secho( - "An error occurred when attempting to refresh your SSO access token. " - "To refresh this session, run 'cloudsmith auth'", - fg="yellow", - err=True, - ) + if context.keyring_refresh_failed and not is_auth_command: if credential: - click.secho( - "Falling back to API key authentication.", - fg="yellow", - err=True, + message = ( + "Using the existing access token until it expires." + if credential.source_name == "keyring" + else "Falling back to alternative authentication." ) + elif context.keyring_refresh_rejected: + message = ( + "Your SSO session has expired. Run 'cloudsmith auth' to " + "authenticate again; continuing without SSO authentication." + ) + elif context.keyring_refresh_unrenewable: + message = ( + "The SSO session has no refresh token and its access token " + "has expired. Run 'cloudsmith auth' to authenticate again; " + "continuing without SSO authentication." + ) + else: + message = ( + "The SSO session could not be renewed and its access token " + "has expired. Check your connection, then run 'cloudsmith auth'; " + "continuing without SSO authentication." + ) + click.secho(message, fg="yellow", err=True) opts.credential = credential diff --git a/cloudsmith_cli/cli/tests/test_decorators.py b/cloudsmith_cli/cli/tests/test_decorators.py index 9f6e5815..b51c3eef 100644 --- a/cloudsmith_cli/cli/tests/test_decorators.py +++ b/cloudsmith_cli/cli/tests/test_decorators.py @@ -1,7 +1,11 @@ +from unittest.mock import patch + import click import click.testing +import pytest -from ..decorators import report_retry +from ...core.credentials.models import CredentialResult +from ..decorators import report_retry, resolve_credentials def test_report_retry_writes_to_stderr(): @@ -14,3 +18,65 @@ def command(): assert result.stdout == '{"data": []}\n' assert "Request was throttled (429)" in result.stderr + + +def _credential_command(name="example"): + @click.command(name=name) + @resolve_credentials + @click.pass_context + def command(ctx, opts): + click.echo("command ran") + + return command + + +def test_rejected_sso_session_continues_anonymously(): + def reject(context): + context.keyring_refresh_failed = True + context.keyring_refresh_rejected = True + return None + + with patch( + "cloudsmith_cli.cli.decorators.CredentialProviderChain.resolve", + side_effect=reject, + ): + result = click.testing.CliRunner().invoke(_credential_command()) + + assert result.exit_code == 0 + assert "Your SSO session has expired" in result.stderr + assert "continuing without SSO authentication" in result.stderr + assert result.stdout == "command ran\n" + + +def test_failed_sso_refresh_reports_alternative_authentication_fallback(): + def fallback(context): + context.keyring_refresh_failed = True + return CredentialResult(api_key="api-key", source_name="oidc") + + with patch( + "cloudsmith_cli.cli.decorators.CredentialProviderChain.resolve", + side_effect=fallback, + ): + result = click.testing.CliRunner().invoke(_credential_command()) + + assert result.exit_code == 0 + assert "Falling back to alternative authentication" in result.stderr + assert result.stdout == "command ran\n" + + +@pytest.mark.parametrize("command_name", ["authenticate", "login"]) +def test_auth_commands_skip_automatic_keyring_refresh(command_name): + def resolve(context): + assert context.skip_keyring_refresh is True + return None + + with patch( + "cloudsmith_cli.cli.decorators.CredentialProviderChain.resolve", + side_effect=resolve, + ): + result = click.testing.CliRunner().invoke( + _credential_command(name=command_name) + ) + + assert result.exit_code == 0 + assert result.stdout == "command ran\n" diff --git a/cloudsmith_cli/core/credentials/models.py b/cloudsmith_cli/core/credentials/models.py index a0ca2020..aefabc24 100644 --- a/cloudsmith_cli/core/credentials/models.py +++ b/cloudsmith_cli/core/credentials/models.py @@ -27,6 +27,9 @@ class CredentialContext: profile: str | None = None debug: bool = False keyring_refresh_failed: bool = False + keyring_refresh_rejected: bool = False + keyring_refresh_unrenewable: bool = False + skip_keyring_refresh: bool = False oidc_audience: str | None = None org: str | None = None oidc_service_slug: str | None = None diff --git a/cloudsmith_cli/core/credentials/providers/keyring_provider.py b/cloudsmith_cli/core/credentials/providers/keyring_provider.py index a4c2f7a4..ee896b1a 100644 --- a/cloudsmith_cli/core/credentials/providers/keyring_provider.py +++ b/cloudsmith_cli/core/credentials/providers/keyring_provider.py @@ -5,36 +5,69 @@ import logging from ....core import keyring -from ...api.exceptions import ApiException -from ...sso import refresh_access_token +from ...sso import SsoRenewalStatus, access_token_is_valid, renew_sso_session from ..models import CredentialContext, CredentialResult from ..provider import CredentialProvider logger = logging.getLogger(__name__) -REFRESH_REJECTED_STATUSES = (400, 401, 403, 422) +def _credential(access_token): + return CredentialResult( + api_key=access_token, + source_name="keyring", + source_detail="SAML token from system keyring", + auth_type="bearer", + ) + + +def _access_token_from_renewal(context, renewal, held_access_token): + if renewal.status == SsoRenewalStatus.REJECTED: + context.keyring_refresh_failed = True + context.keyring_refresh_rejected = True + return None + if renewal.status == SsoRenewalStatus.MISSING: + context.keyring_refresh_failed = True + return held_access_token if access_token_is_valid(held_access_token) else None + if renewal.status == SsoRenewalStatus.FAILED: + context.keyring_refresh_failed = True + return None + if renewal.status == SsoRenewalStatus.UNRENEWABLE: + context.keyring_refresh_failed = True + context.keyring_refresh_unrenewable = True + if not access_token_is_valid(renewal.access_token): + return None + if renewal.status == SsoRenewalStatus.CURRENT and renewal.error: + context.keyring_refresh_failed = True + return renewal.access_token -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) +def _recover_from_unexpected_refresh_error(context, access_token): context.keyring_refresh_failed = True + keyring.update_refresh_attempted_at(context.api_host, profile=context.profile) + return access_token if access_token_is_valid(access_token) else None + + +def _refresh_access_token(context, access_token): + if context.skip_keyring_refresh or not keyring.should_refresh_access_token( + context.api_host, + access_token=access_token, + profile=context.profile, + ): + return access_token + + if not context.session: + logger.debug( + "Session unavailable; skipping token refresh, using existing token" + ) + return access_token + + renewal = renew_sso_session( + context.api_host, + context.session, + profile=context.profile, + ) + return _access_token_from_renewal(context, renewal, access_token) class KeyringProvider(CredentialProvider): @@ -54,52 +87,9 @@ def resolve(self, context: CredentialContext) -> CredentialResult | None: return None try: - 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, 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 + access_token = _refresh_access_token(context, access_token) + except Exception: # pylint: disable=broad-exception-caught + logger.debug("Failed to refresh SAML token", exc_info=True) + access_token = _recover_from_unexpected_refresh_error(context, access_token) - return CredentialResult( - api_key=access_token, - source_name="keyring", - source_detail="SAML token from system keyring", - auth_type="bearer", - ) + return _credential(access_token) if access_token else None diff --git a/cloudsmith_cli/core/keyring.py b/cloudsmith_cli/core/keyring.py index 6919a013..7b247a66 100644 --- a/cloudsmith_cli/core/keyring.py +++ b/cloudsmith_cli/core/keyring.py @@ -16,6 +16,7 @@ def should_use_keyring(): "cloudsmith_cli-access_token_refresh_attempted_at-{api_host}" ) REFRESH_TOKEN_KEY = "cloudsmith_cli-refresh_token-{api_host}" +REFRESH_RETRY_INTERVAL = timedelta(minutes=5) def _get_username(): @@ -182,13 +183,19 @@ def get_access_token(api_host, profile=None): def update_refresh_attempted_at(api_host, refresh_time=None, profile=None): + from keyring.errors import KeyringError + if refresh_time is None: refresh_time = datetime.now(tz=timezone.utc) refresh_attempted_at_value = refresh_time.isoformat() key = _format_key(ACCESS_TOKEN_REFRESH_ATTEMPTED_AT_KEY, api_host, profile) - _set_value(key, refresh_attempted_at_value) + try: + _set_value(key, refresh_attempted_at_value) + except KeyringError: + # This timestamp only throttles retries; it must not block renewal. + pass def get_refresh_attempted_at(api_host, profile=None): @@ -205,10 +212,25 @@ def get_refresh_attempted_at(api_host, profile=None): return None -def should_refresh_access_token(api_host, profile=None): +def should_refresh_access_token(api_host, access_token=None, profile=None): if not should_use_keyring(): return False + if access_token: + from .sso import get_access_token_expiry + + expires_at = get_access_token_expiry(access_token) + if expires_at is not None: + now = datetime.now(tz=timezone.utc) + if expires_at > now + timedelta(minutes=30): + return False + if expires_at <= now: + return True + + attempted_at = get_refresh_attempted_at(api_host, profile=profile) + return not attempted_at or attempted_at < now - REFRESH_RETRY_INTERVAL + + # Preserve the original cadence for opaque tokens without a readable expiry. token_refreshed_at = get_refresh_attempted_at(api_host, profile=profile) if token_refreshed_at: @@ -233,17 +255,19 @@ def store_sso_tokens(api_host, access_token, refresh_token, profile=None): if not should_use_keyring(): return False + # Refresh-token rotation invalidates the old token, so persist its + # replacement before an access-token write can fail. + if refresh_token: + store_refresh_token( + api_host=api_host, refresh_token=refresh_token, profile=profile + ) + if access_token: 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, profile=profile - ) - return True diff --git a/cloudsmith_cli/core/sso.py b/cloudsmith_cli/core/sso.py index a8b4c233..9344089f 100644 --- a/cloudsmith_cli/core/sso.py +++ b/cloudsmith_cli/core/sso.py @@ -1,9 +1,181 @@ """SSO token refresh against the Cloudsmith API.""" +import base64 +import json +import logging +from dataclasses import dataclass +from datetime import datetime, timedelta, timezone +from enum import Enum + import requests +from . import keyring from .api.exceptions import ApiException +logger = logging.getLogger(__name__) +REFRESH_REJECTED_STATUSES = frozenset({400, 401, 403, 422}) +TOKEN_EXPIRY_LEEWAY = timedelta(seconds=30) + + +class SsoRenewalStatus(str, Enum): + """Possible outcomes from renewing an SSO session.""" + + RENEWED = "renewed" + CURRENT = "current" + MISSING = "missing" + UNRENEWABLE = "unrenewable" + REJECTED = "rejected" + FAILED = "failed" + + +@dataclass +class SsoRenewalResult: + """Outcome of renewing the SSO tokens stored for one CLI profile.""" + + status: SsoRenewalStatus + access_token: str | None = None + error: Exception | None = None + + +@dataclass(frozen=True) +class SsoTokens: + """Access and refresh tokens stored for an SSO session.""" + + access_token: str | None + refresh_token: str | None + + +def get_access_token_expiry(access_token): + """Return the expiry encoded in an SSO access token, if available.""" + if not access_token: + return None + + try: + encoded_payload = access_token.split(".", maxsplit=2)[1] + padding = "=" * (-len(encoded_payload) % 4) + payload = json.loads(base64.urlsafe_b64decode(encoded_payload + padding)) + expires_at = payload.get("exp") + if expires_at is not None: + return datetime.fromtimestamp(float(expires_at), tz=timezone.utc) + except ( + IndexError, + OSError, + OverflowError, + TypeError, + ValueError, + json.JSONDecodeError, + ): + logger.debug("Failed to decode SSO access token expiry", exc_info=True) + + return None + + +def access_token_is_valid(access_token, now=None): + """Return whether an SSO access token is present and not provably expired. + + A token without a readable JWT expiry counts as usable. The API stays + the authority on whether it still works. + """ + if not access_token: + return False + + expires_at = get_access_token_expiry(access_token) + if expires_at is None: + return True + + return expires_at + TOKEN_EXPIRY_LEEWAY > (now or datetime.now(tz=timezone.utc)) + + +def _load_sso_tokens(api_host, profile): + return SsoTokens( + access_token=keyring.get_access_token(api_host, profile=profile), + refresh_token=keyring.get_refresh_token(api_host, profile=profile), + ) + + +def _failed_renewal(api_host, profile, access_token, error): + keyring.update_refresh_attempted_at(api_host, profile=profile) + status = ( + SsoRenewalStatus.CURRENT + if access_token_is_valid(access_token) + else SsoRenewalStatus.FAILED + ) + return SsoRenewalResult(status=status, access_token=access_token, error=error) + + +def _recover_from_rejected_renewal(api_host, profile, previous_tokens, error): + current_tokens = _load_sso_tokens(api_host, profile) + if current_tokens != previous_tokens and access_token_is_valid( + current_tokens.access_token + ): + return SsoRenewalResult( + status=SsoRenewalStatus.CURRENT, + access_token=current_tokens.access_token, + ) + + deleted = keyring.delete_sso_tokens(api_host, profile=profile, include_legacy=False) + if not deleted: + keyring.delete_sso_tokens(api_host) + return SsoRenewalResult(status=SsoRenewalStatus.REJECTED, error=error) + + +def _store_renewed_tokens(api_host, profile, access_token, refresh_token): + from keyring.errors import KeyringError + + try: + keyring.store_sso_tokens( + api_host, + access_token, + refresh_token, + profile=profile, + ) + except KeyringError as exc: + return SsoRenewalResult( + status=SsoRenewalStatus.CURRENT, + access_token=access_token, + error=exc, + ) + return SsoRenewalResult( + status=SsoRenewalStatus.RENEWED, + access_token=access_token, + ) + + +def renew_sso_session(api_host, session, profile=None): + """Renew a keyring SSO session and rotate its refresh token.""" + access_token = keyring.get_access_token(api_host, profile=profile) + if not access_token: + return SsoRenewalResult(status=SsoRenewalStatus.MISSING) + + tokens = SsoTokens( + access_token=access_token, + refresh_token=keyring.get_refresh_token(api_host, profile=profile), + ) + if not tokens.refresh_token: + keyring.update_refresh_attempted_at(api_host, profile=profile) + return SsoRenewalResult( + status=SsoRenewalStatus.UNRENEWABLE, + access_token=tokens.access_token, + ) + + try: + new_access_token, new_refresh_token = refresh_access_token( + api_host, + tokens.access_token, + tokens.refresh_token, + session=session, + ) + except (ApiException, requests.RequestException) as exc: + if isinstance(exc, ApiException) and exc.status in REFRESH_REJECTED_STATUSES: + return _recover_from_rejected_renewal(api_host, profile, tokens, error=exc) + return _failed_renewal(api_host, profile, tokens.access_token, error=exc) + + if not new_access_token: + error = ValueError("Cloudsmith did not return a new SSO access token.") + return _failed_renewal(api_host, profile, tokens.access_token, error=error) + + return _store_renewed_tokens(api_host, profile, new_access_token, new_refresh_token) + def raise_for_api_error(response): """Raise :class:`ApiException` if *response* failed, keeping the API's detail. diff --git a/cloudsmith_cli/core/tests/test_keyring.py b/cloudsmith_cli/core/tests/test_keyring.py index a0a1dcc6..f6b60e68 100644 --- a/cloudsmith_cli/core/tests/test_keyring.py +++ b/cloudsmith_cli/core/tests/test_keyring.py @@ -2,8 +2,9 @@ import importlib import os from datetime import datetime, timedelta, timezone -from unittest.mock import ANY, Mock, patch +from unittest.mock import ANY, Mock, call, patch +import jwt import pytest from freezegun import freeze_time from keyrings.cryptfile.cryptfile import CryptFileKeyring @@ -181,6 +182,30 @@ def test_should_refresh_access_token_with_expired_token( "test_user", ) + @freeze_time("2024-06-01 10:00:00") + @pytest.mark.parametrize( + "expires_at,attempted_at,expected", + [ + ("2024-06-01 10:31:00", None, False), + ("2024-06-01 10:30:00", None, True), + ("2024-06-01 09:59:00", "2024-06-01T09:59:00+00:00", True), + ], + ) + def test_jwt_refresh_uses_expiry_and_ignores_throttle_after_expiration( + self, mock_get_password, expires_at, attempted_at, expected + ): + mock_get_password.return_value = attempted_at + access_token = jwt.encode( + {"exp": datetime.fromisoformat(expires_at).replace(tzinfo=timezone.utc)}, + "not-used-for-verification", + algorithm="HS256", + ) + + assert ( + should_refresh_access_token(self.api_host, access_token=access_token) + is expected + ) + def test_store_refresh_token(self, mock_get_user, mock_set_password): store_refresh_token(self.api_host, "refresh_token") @@ -219,25 +244,22 @@ def test_store_sso_tokens(self, mock_get_user, mock_set_password): result = store_sso_tokens(self.api_host, "access_token", "refresh_token") assert result is True - assert mock_set_password.call_count == 3 - mock_set_password.assert_any_call( - "cloudsmith_cli-access_token-https://example.com", - "test_user", - "access_token", - ) refresh_key = ( "cloudsmith_cli-access_token_refresh_attempted_at-https://example.com" ) - mock_set_password.assert_any_call( - refresh_key, - "test_user", - ANY, - ) - mock_set_password.assert_any_call( - "cloudsmith_cli-refresh_token-https://example.com", - "test_user", - "refresh_token", - ) + assert mock_set_password.call_args_list == [ + call( + "cloudsmith_cli-refresh_token-https://example.com", + "test_user", + "refresh_token", + ), + call( + "cloudsmith_cli-access_token-https://example.com", + "test_user", + "access_token", + ), + call(refresh_key, "test_user", ANY), + ] def test_store_sso_tokens_returns_false_when_keyring_disabled( self, mock_get_user, mock_set_password diff --git a/cloudsmith_cli/core/tests/test_keyring_provider.py b/cloudsmith_cli/core/tests/test_keyring_provider.py index 6c1ec47d..9e692884 100644 --- a/cloudsmith_cli/core/tests/test_keyring_provider.py +++ b/cloudsmith_cli/core/tests/test_keyring_provider.py @@ -1,272 +1,67 @@ """Tests for the keyring credential provider.""" import os -from unittest.mock import MagicMock, call, patch - -import pytest +from unittest.mock import MagicMock, patch from cloudsmith_cli.core import keyring -from cloudsmith_cli.core.api.exceptions import ApiException from cloudsmith_cli.core.credentials.models import CredentialContext from cloudsmith_cli.core.credentials.providers import KeyringProvider, keyring_provider - - -@pytest.fixture(autouse=True) -def mock_get_keyring(): - import keyring as keyring_backend - - with patch.object(keyring_backend, "get_keyring") as get_keyring_mock: - yield get_keyring_mock +from cloudsmith_cli.core.sso import SsoRenewalResult, SsoRenewalStatus class TestKeyringProvider: def test_returns_none_when_keyring_disabled(self): - provider = KeyringProvider() with patch.dict(os.environ, {"CLOUDSMITH_NO_KEYRING": "1"}): - result = provider.resolve(CredentialContext()) - assert result is None + assert KeyringProvider().resolve(CredentialContext()) is None def test_returns_none_when_no_token(self): - provider = KeyringProvider() - 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=None), ): - result = provider.resolve(CredentialContext()) - assert result is None + assert KeyringProvider().resolve(CredentialContext()) is None - def test_returns_bearer_token(self): - provider = KeyringProvider() - env = os.environ.copy() - env.pop("CLOUDSMITH_NO_KEYRING", None) + def test_returns_bearer_token_without_refresh(self): 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, "get_access_token", return_value="sso-token"), patch.object(keyring, "should_refresh_access_token", return_value=False), ): - result = provider.resolve(CredentialContext()) - assert result is not None - assert result.api_key == "sso_token" - assert result.auth_type == "bearer" - assert result.source_name == "keyring" - - def test_returns_none_on_refresh_failure(self): - provider = KeyringProvider() - context = CredentialContext(session=MagicMock()) - 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"), - patch.object(keyring, "should_refresh_access_token", return_value=True), - patch.object(keyring, "get_refresh_token", return_value="refresh_tok"), - patch.object( - keyring_provider, - "refresh_access_token", - side_effect=ApiException(status=401, detail="Unauthorized"), - ), - patch.object(keyring, "delete_sso_tokens", return_value=True), - patch.object(keyring, "update_refresh_attempted_at"), - ): - result = provider.resolve(context) - 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) + result = KeyringProvider().resolve(CredentialContext()) 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.api_key, result.auth_type) == ("sso-token", "bearer") - 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 + def test_transient_failure_retains_usable_token(self): + context = CredentialContext(session=MagicMock()) + renewal = SsoRenewalResult( + status=SsoRenewalStatus.CURRENT, + access_token="still-usable", + error=ConnectionError("offline"), ) - 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, "get_access_token", return_value="old-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, + patch.object(keyring_provider, "renew_sso_session", return_value=renewal), ): - 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) + result = KeyringProvider().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 result.api_key == "still-usable" 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) + def test_rejected_session_is_not_returned(self): + context = CredentialContext(session=MagicMock()) + renewal = SsoRenewalResult(status=SsoRenewalStatus.REJECTED) 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, "get_access_token", return_value="dead-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, + patch.object(keyring_provider, "renew_sso_session", return_value=renewal), ): - result = provider.resolve(context) + result = KeyringProvider().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") + assert context.keyring_refresh_rejected is True diff --git a/cloudsmith_cli/core/tests/test_sso.py b/cloudsmith_cli/core/tests/test_sso.py new file mode 100644 index 00000000..95a67638 --- /dev/null +++ b/cloudsmith_cli/core/tests/test_sso.py @@ -0,0 +1,117 @@ +"""Tests for shared SSO session renewal.""" + +from datetime import datetime, timezone +from unittest.mock import MagicMock, call, patch + +import jwt +import requests +from freezegun import freeze_time + +from cloudsmith_cli.core.api.exceptions import ApiException +from cloudsmith_cli.core.sso import renew_sso_session + +API_HOST = "https://api.example.com" + + +def test_successful_renewal_stores_rotated_tokens(): + with ( + patch( + "cloudsmith_cli.core.sso.keyring.get_access_token", + return_value="old-access", + ), + patch( + "cloudsmith_cli.core.sso.keyring.get_refresh_token", + return_value="old-refresh", + ), + patch( + "cloudsmith_cli.core.sso.refresh_access_token", + return_value=("new-access", "new-refresh"), + ), + patch("cloudsmith_cli.core.sso.keyring.store_sso_tokens") as store, + ): + result = renew_sso_session(API_HOST, MagicMock(), profile="work") + + assert (result.status, result.access_token) == ("renewed", "new-access") + store.assert_called_once_with(API_HOST, "new-access", "new-refresh", profile="work") + + +@freeze_time("2024-06-01 10:00:00") +def test_transient_failure_reuses_usable_access_token(): + access_token = jwt.encode( + {"exp": datetime(2024, 6, 1, 9, 59, 31, tzinfo=timezone.utc)}, + "not-used-for-verification", + algorithm="HS256", + ) + error = requests.ConnectionError("offline") + with ( + patch( + "cloudsmith_cli.core.sso.keyring.get_access_token", + return_value=access_token, + ), + patch( + "cloudsmith_cli.core.sso.keyring.get_refresh_token", + return_value="old-refresh", + ), + patch("cloudsmith_cli.core.sso.refresh_access_token", side_effect=error), + patch( + "cloudsmith_cli.core.sso.keyring.update_refresh_attempted_at" + ) as attempted, + ): + result = renew_sso_session(API_HOST, MagicMock(), profile="work") + + assert (result.status, result.access_token, result.error) == ( + "current", + access_token, + error, + ) + attempted.assert_called_once_with(API_HOST, profile="work") + + +def test_rejected_renewal_reuses_concurrently_rotated_tokens(): + with ( + patch( + "cloudsmith_cli.core.sso.keyring.get_access_token", + side_effect=["old-access", "new-access"], + ), + patch( + "cloudsmith_cli.core.sso.keyring.get_refresh_token", + side_effect=["old-refresh", "new-refresh"], + ), + patch( + "cloudsmith_cli.core.sso.refresh_access_token", + side_effect=ApiException(400, detail="Already rotated"), + ), + patch("cloudsmith_cli.core.sso.keyring.delete_sso_tokens") as delete, + ): + result = renew_sso_session(API_HOST, MagicMock()) + + assert (result.status, result.access_token) == ("current", "new-access") + delete.assert_not_called() + + +def test_definitive_rejection_cleans_up_profile_then_legacy_tokens(): + with ( + patch( + "cloudsmith_cli.core.sso.keyring.get_access_token", + return_value="old-access", + ), + patch( + "cloudsmith_cli.core.sso.keyring.get_refresh_token", + return_value="old-refresh", + ), + patch( + "cloudsmith_cli.core.sso.refresh_access_token", + side_effect=ApiException(401, detail="Rejected"), + ), + patch( + "cloudsmith_cli.core.sso.keyring.delete_sso_tokens", + side_effect=[False, True], + ) as delete, + ): + result = renew_sso_session(API_HOST, MagicMock(), profile="work") + + assert result.status == "rejected" + assert delete.call_args_list == [ + call(API_HOST, profile="work", include_legacy=False), + call(API_HOST), + ] From 48a08909286dead2c5d5b7c0d321b32da7bbfbda Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Thu, 3 Sep 2026 00:21:08 +0100 Subject: [PATCH 2/3] Refine SSO renewal test coverage --- cloudsmith_cli/cli/tests/test_decorators.py | 17 -- .../core/tests/test_keyring_provider.py | 178 +++++++++++++++--- 2 files changed, 155 insertions(+), 40 deletions(-) diff --git a/cloudsmith_cli/cli/tests/test_decorators.py b/cloudsmith_cli/cli/tests/test_decorators.py index b51c3eef..b83bc01d 100644 --- a/cloudsmith_cli/cli/tests/test_decorators.py +++ b/cloudsmith_cli/cli/tests/test_decorators.py @@ -4,7 +4,6 @@ import click.testing import pytest -from ...core.credentials.models import CredentialResult from ..decorators import report_retry, resolve_credentials @@ -48,22 +47,6 @@ def reject(context): assert result.stdout == "command ran\n" -def test_failed_sso_refresh_reports_alternative_authentication_fallback(): - def fallback(context): - context.keyring_refresh_failed = True - return CredentialResult(api_key="api-key", source_name="oidc") - - with patch( - "cloudsmith_cli.cli.decorators.CredentialProviderChain.resolve", - side_effect=fallback, - ): - result = click.testing.CliRunner().invoke(_credential_command()) - - assert result.exit_code == 0 - assert "Falling back to alternative authentication" in result.stderr - assert result.stdout == "command ran\n" - - @pytest.mark.parametrize("command_name", ["authenticate", "login"]) def test_auth_commands_skip_automatic_keyring_refresh(command_name): def resolve(context): diff --git a/cloudsmith_cli/core/tests/test_keyring_provider.py b/cloudsmith_cli/core/tests/test_keyring_provider.py index 9e692884..e5e52cff 100644 --- a/cloudsmith_cli/core/tests/test_keyring_provider.py +++ b/cloudsmith_cli/core/tests/test_keyring_provider.py @@ -4,54 +4,113 @@ from unittest.mock import MagicMock, patch from cloudsmith_cli.core import keyring -from cloudsmith_cli.core.credentials.models import CredentialContext +from cloudsmith_cli.core.credentials.chain import CredentialProviderChain +from cloudsmith_cli.core.credentials.models import CredentialContext, CredentialResult from cloudsmith_cli.core.credentials.providers import KeyringProvider, keyring_provider +from cloudsmith_cli.core.credentials.provider import CredentialProvider from cloudsmith_cli.core.sso import SsoRenewalResult, SsoRenewalStatus class TestKeyringProvider: def test_returns_none_when_keyring_disabled(self): + provider = KeyringProvider() with patch.dict(os.environ, {"CLOUDSMITH_NO_KEYRING": "1"}): - assert KeyringProvider().resolve(CredentialContext()) is None + result = provider.resolve(CredentialContext()) + assert result is None def test_returns_none_when_no_token(self): + provider = KeyringProvider() + 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=None), ): - assert KeyringProvider().resolve(CredentialContext()) is None + result = provider.resolve(CredentialContext()) + assert result is None - def test_returns_bearer_token_without_refresh(self): + def test_returns_bearer_token(self): + provider = KeyringProvider() + 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, "get_access_token", return_value="sso_token"), patch.object(keyring, "should_refresh_access_token", return_value=False), ): - result = KeyringProvider().resolve(CredentialContext()) + result = provider.resolve(CredentialContext()) + assert result is not None + assert result.api_key == "sso_token" + assert result.auth_type == "bearer" + assert result.source_name == "keyring" - assert result is not None - assert (result.api_key, result.auth_type) == ("sso-token", "bearer") - - def test_transient_failure_retains_usable_token(self): + def test_returns_none_on_refresh_failure(self): + provider = KeyringProvider() context = CredentialContext(session=MagicMock()) - renewal = SsoRenewalResult( - status=SsoRenewalStatus.CURRENT, - access_token="still-usable", - error=ConnectionError("offline"), - ) + 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"), + patch.object(keyring, "get_access_token", return_value="old_token"), patch.object(keyring, "should_refresh_access_token", return_value=True), - patch.object(keyring_provider, "renew_sso_session", return_value=renewal), + patch.object( + keyring_provider, + "renew_sso_session", + return_value=SsoRenewalResult(status=SsoRenewalStatus.FAILED), + ), ): - result = KeyringProvider().resolve(context) + 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_provider, + "renew_sso_session", + return_value=SsoRenewalResult( + status=SsoRenewalStatus.RENEWED, + access_token="new_token", + ), + ) as renew_mock, + ): + result = provider.resolve(context) assert result is not None - assert result.api_key == "still-usable" - assert context.keyring_refresh_failed is True + 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, access_token="old_token", profile="staging" + ) + renew_mock.assert_called_once_with( + context.api_host, context.session, profile="staging" + ) + + def test_rejected_session_continues_down_provider_chain(self): + class FallbackProvider(CredentialProvider): + name = "fallback" + + def resolve(self, context): + return CredentialResult( + api_key="fallback-credential", + source_name=self.name, + ) - def test_rejected_session_is_not_returned(self): context = CredentialContext(session=MagicMock()) renewal = SsoRenewalResult(status=SsoRenewalStatus.REJECTED) with ( @@ -60,8 +119,81 @@ def test_rejected_session_is_not_returned(self): patch.object(keyring, "should_refresh_access_token", return_value=True), patch.object(keyring_provider, "renew_sso_session", return_value=renewal), ): - result = KeyringProvider().resolve(context) + result = CredentialProviderChain( + [KeyringProvider(), FallbackProvider()] + ).resolve(context) + + assert result.api_key == "fallback-credential" + assert context.keyring_refresh_rejected is True + + 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_provider, + "renew_sso_session", + return_value=SsoRenewalResult( + status=SsoRenewalStatus.UNRENEWABLE, + access_token="sso_token", + ), + ), + ): + result = provider.resolve(context) + + assert result is not None + assert result.api_key == "sso_token" + assert context.keyring_refresh_unrenewable is True + + 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_provider, + "renew_sso_session", + return_value=SsoRenewalResult( + status=SsoRenewalStatus.CURRENT, + access_token="stale_token", + error=ConnectionError("offline"), + ), + ), + ): + result = provider.resolve(context) + + assert result is not None + assert result.api_key == "stale_token" + assert context.keyring_refresh_failed is True + + 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_provider, + "renew_sso_session", + return_value=SsoRenewalResult(status=SsoRenewalStatus.FAILED), + ), + ): + result = provider.resolve(context) assert result is None assert context.keyring_refresh_failed is True - assert context.keyring_refresh_rejected is True From fd9a17aaf1a81f422301a898346c47f05febbbf2 Mon Sep 17 00:00:00 2001 From: Ian Duffy Date: Thu, 3 Sep 2026 00:22:13 +0100 Subject: [PATCH 3/3] Limit SSO layer to renewal primitives --- cloudsmith_cli/cli/decorators.py | 36 ++-- cloudsmith_cli/cli/tests/test_decorators.py | 51 +---- cloudsmith_cli/core/credentials/models.py | 3 - .../credentials/providers/keyring_provider.py | 128 +++++++------ .../core/tests/test_keyring_provider.py | 179 ++++++++++++------ 5 files changed, 207 insertions(+), 190 deletions(-) diff --git a/cloudsmith_cli/cli/decorators.py b/cloudsmith_cli/cli/decorators.py index f5124cfb..a76305e8 100644 --- a/cloudsmith_cli/cli/decorators.py +++ b/cloudsmith_cli/cli/decorators.py @@ -495,7 +495,6 @@ def wrapper(ctx, *args, **kwargs): opts.oidc_detector_order, oidc_disabled_detectors ) - is_auth_command = ctx.command.name in ("authenticate", "login") context = CredentialContext( session=opts.session, api_key_from_flag=opts.api_key_from_flag, @@ -511,37 +510,24 @@ def wrapper(ctx, *args, **kwargs): oidc_discovery_disabled=opts.oidc_discovery_disabled, oidc_detector_order=opts.oidc_detector_order, oidc_disabled_detectors=oidc_disabled_detectors, - skip_keyring_refresh=is_auth_command, ) chain = CredentialProviderChain() credential = chain.resolve(context) - if context.keyring_refresh_failed and not is_auth_command: + if context.keyring_refresh_failed: + click.secho( + "An error occurred when attempting to refresh your SSO access token. " + "To refresh this session, run 'cloudsmith auth'", + fg="yellow", + err=True, + ) if credential: - message = ( - "Using the existing access token until it expires." - if credential.source_name == "keyring" - else "Falling back to alternative authentication." + click.secho( + "Falling back to API key authentication.", + fg="yellow", + err=True, ) - elif context.keyring_refresh_rejected: - message = ( - "Your SSO session has expired. Run 'cloudsmith auth' to " - "authenticate again; continuing without SSO authentication." - ) - elif context.keyring_refresh_unrenewable: - message = ( - "The SSO session has no refresh token and its access token " - "has expired. Run 'cloudsmith auth' to authenticate again; " - "continuing without SSO authentication." - ) - else: - message = ( - "The SSO session could not be renewed and its access token " - "has expired. Check your connection, then run 'cloudsmith auth'; " - "continuing without SSO authentication." - ) - click.secho(message, fg="yellow", err=True) opts.credential = credential diff --git a/cloudsmith_cli/cli/tests/test_decorators.py b/cloudsmith_cli/cli/tests/test_decorators.py index b83bc01d..9f6e5815 100644 --- a/cloudsmith_cli/cli/tests/test_decorators.py +++ b/cloudsmith_cli/cli/tests/test_decorators.py @@ -1,10 +1,7 @@ -from unittest.mock import patch - import click import click.testing -import pytest -from ..decorators import report_retry, resolve_credentials +from ..decorators import report_retry def test_report_retry_writes_to_stderr(): @@ -17,49 +14,3 @@ def command(): assert result.stdout == '{"data": []}\n' assert "Request was throttled (429)" in result.stderr - - -def _credential_command(name="example"): - @click.command(name=name) - @resolve_credentials - @click.pass_context - def command(ctx, opts): - click.echo("command ran") - - return command - - -def test_rejected_sso_session_continues_anonymously(): - def reject(context): - context.keyring_refresh_failed = True - context.keyring_refresh_rejected = True - return None - - with patch( - "cloudsmith_cli.cli.decorators.CredentialProviderChain.resolve", - side_effect=reject, - ): - result = click.testing.CliRunner().invoke(_credential_command()) - - assert result.exit_code == 0 - assert "Your SSO session has expired" in result.stderr - assert "continuing without SSO authentication" in result.stderr - assert result.stdout == "command ran\n" - - -@pytest.mark.parametrize("command_name", ["authenticate", "login"]) -def test_auth_commands_skip_automatic_keyring_refresh(command_name): - def resolve(context): - assert context.skip_keyring_refresh is True - return None - - with patch( - "cloudsmith_cli.cli.decorators.CredentialProviderChain.resolve", - side_effect=resolve, - ): - result = click.testing.CliRunner().invoke( - _credential_command(name=command_name) - ) - - assert result.exit_code == 0 - assert result.stdout == "command ran\n" diff --git a/cloudsmith_cli/core/credentials/models.py b/cloudsmith_cli/core/credentials/models.py index aefabc24..a0ca2020 100644 --- a/cloudsmith_cli/core/credentials/models.py +++ b/cloudsmith_cli/core/credentials/models.py @@ -27,9 +27,6 @@ class CredentialContext: profile: str | None = None debug: bool = False keyring_refresh_failed: bool = False - keyring_refresh_rejected: bool = False - keyring_refresh_unrenewable: bool = False - skip_keyring_refresh: bool = False oidc_audience: str | None = None org: str | None = None oidc_service_slug: str | None = None diff --git a/cloudsmith_cli/core/credentials/providers/keyring_provider.py b/cloudsmith_cli/core/credentials/providers/keyring_provider.py index ee896b1a..a4c2f7a4 100644 --- a/cloudsmith_cli/core/credentials/providers/keyring_provider.py +++ b/cloudsmith_cli/core/credentials/providers/keyring_provider.py @@ -5,69 +5,36 @@ import logging from ....core import keyring -from ...sso import SsoRenewalStatus, access_token_is_valid, renew_sso_session +from ...api.exceptions import ApiException +from ...sso import refresh_access_token from ..models import CredentialContext, CredentialResult from ..provider import CredentialProvider logger = logging.getLogger(__name__) - -def _credential(access_token): - return CredentialResult( - api_key=access_token, - source_name="keyring", - source_detail="SAML token from system keyring", - auth_type="bearer", - ) - - -def _access_token_from_renewal(context, renewal, held_access_token): - if renewal.status == SsoRenewalStatus.REJECTED: - context.keyring_refresh_failed = True - context.keyring_refresh_rejected = True - return None - if renewal.status == SsoRenewalStatus.MISSING: - context.keyring_refresh_failed = True - return held_access_token if access_token_is_valid(held_access_token) else None - if renewal.status == SsoRenewalStatus.FAILED: - context.keyring_refresh_failed = True - return None - if renewal.status == SsoRenewalStatus.UNRENEWABLE: - context.keyring_refresh_failed = True - context.keyring_refresh_unrenewable = True - if not access_token_is_valid(renewal.access_token): - return None - if renewal.status == SsoRenewalStatus.CURRENT and renewal.error: - context.keyring_refresh_failed = True - return renewal.access_token - - -def _recover_from_unexpected_refresh_error(context, access_token): - context.keyring_refresh_failed = True - keyring.update_refresh_attempted_at(context.api_host, profile=context.profile) - return access_token if access_token_is_valid(access_token) else None +REFRESH_REJECTED_STATUSES = (400, 401, 403, 422) -def _refresh_access_token(context, access_token): - if context.skip_keyring_refresh or not keyring.should_refresh_access_token( - context.api_host, - access_token=access_token, - profile=context.profile, - ): - return access_token +def _handle_refresh_failure(context, wipe_tokens): + """Record a refresh failure and clear rejected tokens. - if not context.session: - logger.debug( - "Session unavailable; skipping token refresh, using existing token" + 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 ) - return access_token - - renewal = renew_sso_session( - context.api_host, - context.session, - profile=context.profile, - ) - return _access_token_from_renewal(context, renewal, access_token) + 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): @@ -87,9 +54,52 @@ def resolve(self, context: CredentialContext) -> CredentialResult | None: return None try: - access_token = _refresh_access_token(context, access_token) - except Exception: # pylint: disable=broad-exception-caught - logger.debug("Failed to refresh SAML token", exc_info=True) - access_token = _recover_from_unexpected_refresh_error(context, access_token) + 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, 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 _credential(access_token) if access_token else None + return CredentialResult( + api_key=access_token, + source_name="keyring", + source_detail="SAML token from system keyring", + auth_type="bearer", + ) diff --git a/cloudsmith_cli/core/tests/test_keyring_provider.py b/cloudsmith_cli/core/tests/test_keyring_provider.py index e5e52cff..6c1ec47d 100644 --- a/cloudsmith_cli/core/tests/test_keyring_provider.py +++ b/cloudsmith_cli/core/tests/test_keyring_provider.py @@ -1,14 +1,22 @@ """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.core import keyring -from cloudsmith_cli.core.credentials.chain import CredentialProviderChain -from cloudsmith_cli.core.credentials.models import CredentialContext, CredentialResult +from cloudsmith_cli.core.api.exceptions import ApiException +from cloudsmith_cli.core.credentials.models import CredentialContext from cloudsmith_cli.core.credentials.providers import KeyringProvider, keyring_provider -from cloudsmith_cli.core.credentials.provider import CredentialProvider -from cloudsmith_cli.core.sso import SsoRenewalResult, SsoRenewalStatus + + +@pytest.fixture(autouse=True) +def mock_get_keyring(): + import keyring as keyring_backend + + with patch.object(keyring_backend, "get_keyring") as get_keyring_mock: + yield get_keyring_mock class TestKeyringProvider: @@ -56,11 +64,14 @@ def test_returns_none_on_refresh_failure(self): patch.object(keyring, "should_use_keyring", return_value=True), patch.object(keyring, "get_access_token", return_value="old_token"), patch.object(keyring, "should_refresh_access_token", return_value=True), + patch.object(keyring, "get_refresh_token", return_value="refresh_tok"), patch.object( keyring_provider, - "renew_sso_session", - return_value=SsoRenewalResult(status=SsoRenewalStatus.FAILED), + "refresh_access_token", + side_effect=ApiException(status=401, detail="Unauthorized"), ), + patch.object(keyring, "delete_sso_tokens", return_value=True), + patch.object(keyring, "update_refresh_attempted_at"), ): result = provider.resolve(context) assert result is None @@ -80,53 +91,88 @@ def test_passes_profile_to_keyring(self): 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, - "renew_sso_session", - return_value=SsoRenewalResult( - status=SsoRenewalStatus.RENEWED, - access_token="new_token", - ), - ) as renew_mock, + "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, access_token="old_token", profile="staging" - ) - renew_mock.assert_called_once_with( - context.api_host, context.session, 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" ) - def test_rejected_session_continues_down_provider_chain(self): - class FallbackProvider(CredentialProvider): - name = "fallback" + @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) - def resolve(self, context): - return CredentialResult( - api_key="fallback-credential", - source_name=self.name, - ) + 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() - context = CredentialContext(session=MagicMock()) - renewal = SsoRenewalResult(status=SsoRenewalStatus.REJECTED) + 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="dead-token"), + patch.object(keyring, "get_access_token", return_value="stale_token"), patch.object(keyring, "should_refresh_access_token", return_value=True), - patch.object(keyring_provider, "renew_sso_session", return_value=renewal), + 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 = CredentialProviderChain( - [KeyringProvider(), FallbackProvider()] - ).resolve(context) + result = provider.resolve(context) - assert result.api_key == "fallback-credential" - assert context.keyring_refresh_rejected is True + 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_skips_refresh_when_no_refresh_token_is_stored(self): + def test_stamps_attempt_when_wipe_removes_nothing(self): provider = KeyringProvider() context = CredentialContext(session=MagicMock(), profile="staging") env = os.environ.copy() @@ -134,22 +180,44 @@ def test_skips_refresh_when_no_refresh_token_is_stored(self): 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, "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, - "renew_sso_session", - return_value=SsoRenewalResult( - status=SsoRenewalStatus.UNRENEWABLE, - access_token="sso_token", - ), + "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_unrenewable is True + 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() @@ -161,21 +229,21 @@ def test_keeps_tokens_on_transient_refresh_error(self): 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, - "renew_sso_session", - return_value=SsoRenewalResult( - status=SsoRenewalStatus.CURRENT, - access_token="stale_token", - error=ConnectionError("offline"), - ), + "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 not None - assert result.api_key == "stale_token" + 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() @@ -187,13 +255,18 @@ def test_refresh_without_access_token_in_response_is_a_failure(self): 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, - "renew_sso_session", - return_value=SsoRenewalResult(status=SsoRenewalStatus.FAILED), + "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")