Harden automatic SSO renewal

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

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
 
 

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.

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