Integrate SSO renewal with credential resolution

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
 

cloudsmith_cli/cli/tests/test_decorators.py

@@ -1,7 +1,10 @@
+from unittest.mock import patch
+
 import click
 import click.testing
+import pytest
 
-from ..decorators import report_retry
+from ..decorators import report_retry, resolve_credentials
 
 
 def test_report_retry_writes_to_stderr():
@@ -14,3 +17,49 @@ 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
+    def command(opts):
+        click.echo("command ran")
+
+    return command
+
+
+def test_rejected_sso_session_continues_without_early_exception():
+    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"
+    assert result.stderr == ""

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

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