feat: add credential provider chain concept

Author: cloudsmith-iduffy

cloudsmith_cli/cli/decorators.py

@@ -7,6 +7,7 @@
 from cloudsmith_cli.cli import validators
 
 from ..core.api.init import initialise_api as _initialise_api
+from ..core.credentials import CredentialContext, CredentialProviderChain
 from ..core.mcp import server
 from . import config, utils
 
@@ -314,13 +315,51 @@ def _set_boolean(name, invert=False):
         opts.error_retry_codes = kwargs.pop("error_retry_codes")
         opts.error_retry_cb = report_retry
 
+        credential_context = CredentialContext(
+            api_host=opts.api_host or "https://api.cloudsmith.io",
+            creds_file_path=ctx.meta.get("creds_file"),
+            profile=ctx.meta.get("profile"),
+            debug=opts.debug,
+            cli_api_key=opts.api_key,
+            proxy=opts.api_proxy,
+            ssl_verify=opts.api_ssl_verify if opts.api_ssl_verify is not None else True,
+            user_agent=opts.api_user_agent,
+            headers=opts.api_headers,
+        )
+
+        chain = CredentialProviderChain()
+
+        credential = chain.resolve(credential_context)
+
+        if credential_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:
+                click.secho(
+                    "Falling back to API key authentication.",
+                    fg="yellow",
+                    err=True,
+                )
+
+        resolved_key = None
+        resolved_access_token = None
+        if credential:
+            if credential.auth_type == "bearer":
+                resolved_access_token = credential.api_key
+            else:
+                resolved_key = credential.api_key
+
         def call_print_rate_limit_info_with_opts(rate_info):
             utils.print_rate_limit_info(opts, rate_info)
 
         opts.api_config = _initialise_api(
             debug=opts.debug,
             host=opts.api_host,
-            key=opts.api_key,
+            key=resolved_key,
             proxy=opts.api_proxy,
             ssl_verify=opts.api_ssl_verify,
             user_agent=opts.api_user_agent,
@@ -331,6 +370,7 @@ def call_print_rate_limit_info_with_opts(rate_info):
             error_retry_backoff=opts.error_retry_backoff,
             error_retry_codes=opts.error_retry_codes,
             error_retry_cb=opts.error_retry_cb,
+            access_token=resolved_access_token,
         )
 
         kwargs["opts"] = opts

cloudsmith_cli/core/api/init.py

@@ -6,10 +6,7 @@
 import click
 import cloudsmith_api
 
-from ...cli import saml
-from .. import keyring
 from ..rest import RestClient
-from .exceptions import ApiException
 
 
 def initialise_api(
@@ -45,63 +42,14 @@ def initialise_api(
     config.verify_ssl = ssl_verify
     config.client_side_validation = False
 
-    # Use directly provided access token (e.g. from SSO callback),
-    # or fall back to keyring lookup if enabled.
-    if not access_token:
-        access_token = keyring.get_access_token(config.host)
-
     if access_token:
-        auth_header = config.headers.get("Authorization")
-
-        # overwrite auth header if empty or is basic auth without username or password
-        if not auth_header or auth_header == config.get_basic_auth_token():
-            refresh_token = keyring.get_refresh_token(config.host)
-
-            try:
-                if keyring.should_refresh_access_token(config.host):
-                    new_access_token, new_refresh_token = saml.refresh_access_token(
-                        config.host,
-                        access_token,
-                        refresh_token,
-                        session=saml.create_configured_session(config),
-                    )
-                    keyring.store_sso_tokens(
-                        config.host, new_access_token, new_refresh_token
-                    )
-                    # Use the new tokens
-                    access_token = new_access_token
-            except ApiException:
-                keyring.update_refresh_attempted_at(config.host)
-
-                click.secho(
-                    "An error occurred when attempting to refresh your SSO access token. To refresh this session, run 'cloudsmith auth'",
-                    fg="yellow",
-                    err=True,
-                )
-
-                # Clear access_token to prevent using expired token
-                access_token = None
-
-                # Fall back to API key auth if available
-                if key:
-                    click.secho(
-                        "Falling back to API key authentication.",
-                        fg="yellow",
-                        err=True,
-                    )
-                    config.api_key["X-Api-Key"] = key
-
-            # Only use SSO token if refresh didn't fail
-            if access_token:
-                config.headers["Authorization"] = "Bearer {access_token}".format(
-                    access_token=access_token
-                )
-
-                if config.debug:
-                    click.echo("SSO access token config value set")
+        config.headers["Authorization"] = "Bearer {access_token}".format(
+            access_token=access_token
+        )
+        if config.debug:
+            click.echo("SSO access token config value set")
     elif key:
         config.api_key["X-Api-Key"] = key
-
         if config.debug:
             click.echo("User API key config value set")
 

cloudsmith_cli/core/credentials/__init__.py

@@ -0,0 +1,106 @@
+"""Credential Provider Chain for Cloudsmith CLI.
+
+Implements an AWS SDK-style credential resolution chain that evaluates
+credential sources sequentially and returns the first valid result.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class CredentialContext:  # pylint: disable=too-many-instance-attributes
+    """Context passed to credential providers during resolution."""
+
+    api_host: str = "https://api.cloudsmith.io"
+    config_file_path: str | None = None
+    creds_file_path: str | None = None
+    profile: str | None = None
+    debug: bool = False
+    # Pre-resolved values from CLI flags (highest priority)
+    cli_api_key: str | None = None
+    # API networking configuration
+    proxy: str | None = None
+    ssl_verify: bool = True
+    user_agent: str | None = None
+    headers: dict | None = None
+    keyring_refresh_failed: bool = False
+
+
+@dataclass
+class CredentialResult:
+    """Result from a successful credential resolution."""
+
+    api_key: str
+    source_name: str
+    source_detail: str | None = None
+    auth_type: str = "api_key"
+
+
+class CredentialProvider:
+    """Base class for credential providers."""
+
+    name: str = "base"
+
+    def resolve(self, context: CredentialContext) -> CredentialResult | None:
+        """Attempt to resolve credentials. Return CredentialResult or None."""
+        raise NotImplementedError
+
+
+class CredentialProviderChain:
+    """Evaluates credential providers in order, returning the first valid result.
+
+    If no providers are given, uses the default chain:
+    Keyring → CLIFlag → EnvironmentVariable → ConfigFile.
+    """
+
+    def __init__(self, providers: list[CredentialProvider] | None = None):
+        if providers is not None:
+            self.providers = providers
+        else:
+            from .providers import (
+                CLIFlagProvider,
+                ConfigFileProvider,
+                EnvironmentVariableProvider,
+                KeyringProvider,
+            )
+
+            self.providers = [
+                KeyringProvider(),
+                CLIFlagProvider(),
+                EnvironmentVariableProvider(),
+                ConfigFileProvider(),
+            ]
+
+    def resolve(self, context: CredentialContext) -> CredentialResult | None:
+        """Evaluate each provider in order. Return the first successful result."""
+        for provider in self.providers:
+            try:
+                result = provider.resolve(context)
+                if result is not None:
+                    if context.debug:
+                        logger.debug(
+                            "Credentials resolved by %s: %s",
+                            provider.name,
+                            result.source_detail or result.source_name,
+                        )
+                    return result
+                if context.debug:
+                    logger.debug(
+                        "Provider %s did not resolve credentials, trying next",
+                        provider.name,
+                    )
+            except Exception:  # pylint: disable=broad-exception-caught
+                # Intentionally broad - one provider failing shouldn't stop others
+                logger.debug(
+                    "Provider %s raised an exception, skipping",
+                    provider.name,
+                    exc_info=True,
+                )
+                continue
+        return None

cloudsmith_cli/core/credentials/providers/__init__.py

@@ -0,0 +1,13 @@
+"""Credential providers for the Cloudsmith CLI."""
+
+from .cli_flag import CLIFlagProvider
+from .config_file import ConfigFileProvider
+from .environment_variable import EnvironmentVariableProvider
+from .keyring_provider import KeyringProvider
+
+__all__ = [
+    "CLIFlagProvider",
+    "ConfigFileProvider",
+    "EnvironmentVariableProvider",
+    "KeyringProvider",
+]

cloudsmith_cli/core/credentials/providers/cli_flag.py

@@ -0,0 +1,22 @@
+"""CLI flag credential provider."""
+
+from __future__ import annotations
+
+from .. import CredentialContext, CredentialProvider, CredentialResult
+
+
+class CLIFlagProvider(CredentialProvider):
+    """Resolves credentials from a CLI flag value passed via CredentialContext."""
+
+    name = "cli_flag"
+
+    def resolve(self, context: CredentialContext) -> CredentialResult | None:
+        api_key = context.cli_api_key
+        if api_key and api_key.strip():
+            suffix = api_key.strip()[-4:]
+            return CredentialResult(
+                api_key=api_key.strip(),
+                source_name="cli_flag",
+                source_detail=f"--api-key flag or CLOUDSMITH_API_KEY (ends with ...{suffix})",
+            )
+        return None

cloudsmith_cli/core/credentials/providers/config_file.py

@@ -0,0 +1,55 @@
+"""Config file credential provider."""
+
+from __future__ import annotations
+
+import logging
+import os
+
+from .. import CredentialContext, CredentialProvider, CredentialResult
+
+logger = logging.getLogger(__name__)
+
+
+class ConfigFileProvider(CredentialProvider):
+    """Resolves credentials from the credentials.ini config file."""
+
+    name = "config_file"
+
+    def resolve(self, context: CredentialContext) -> CredentialResult | None:
+        from ....cli.config import CredentialsReader
+
+        reader = CredentialsReader
+        path = context.creds_file_path
+
+        try:
+            config = {}
+            if path and os.path.exists(path):
+                if os.path.isdir(path):
+                    reader.config_searchpath.insert(0, path)
+                else:
+                    reader.config_files.insert(0, path)
+
+            raw_config = reader.read_config()
+            values = raw_config.get("default", {})
+            config.update(values)
+
+            if context.profile and context.profile != "default":
+                profile_values = raw_config.get(f"profile:{context.profile}", {})
+                config.update(profile_values)
+
+            api_key = config.get("api_key")
+            if api_key and isinstance(api_key, str):
+                api_key = api_key.strip().strip("'\"")
+                if api_key:
+                    source_files = reader.find_existing_files()
+                    source = source_files[0] if source_files else "credentials.ini"
+                    return CredentialResult(
+                        api_key=api_key,
+                        source_name="config_file",
+                        source_detail=f"credentials.ini ({source})",
+                    )
+        except Exception:  # pylint: disable=broad-exception-caught
+            # Config file errors can be varied (permissions, parse errors, etc.)
+            logger.debug("ConfigFileProvider failed to read config", exc_info=True)
+
+        return None

cloudsmith_cli/core/credentials/providers/environment_variable.py

@@ -0,0 +1,24 @@
+"""Environment variable credential provider."""
+
+from __future__ import annotations
+
+import os
+
+from .. import CredentialContext, CredentialProvider, CredentialResult
+
+
+class EnvironmentVariableProvider(CredentialProvider):
+    """Resolves credentials from the CLOUDSMITH_API_KEY environment variable."""
+
+    name = "environment_variable"
+
+    def resolve(self, context: CredentialContext) -> CredentialResult | None:
+        api_key = os.environ.get("CLOUDSMITH_API_KEY")
+        if api_key and api_key.strip():
+            suffix = api_key.strip()[-4:]
+            return CredentialResult(
+                api_key=api_key.strip(),
+                source_name="environment_variable",
+                source_detail=f"CLOUDSMITH_API_KEY env var (ends with ...{suffix})",
+            )
+        return None