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/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_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), + ]