feat: attempt to seperate config/credential resolution from the cli

Author: cloudsmith-iduffy

cloudsmith_cli/cli/commands/auth.py

@@ -4,9 +4,10 @@
 
 import click
 
+from ...core.credentials.session import create_session
+from ...core.saml import get_idp_url
 from .. import decorators, utils, validators
 from ..exceptions import handle_api_exceptions
-from ..saml import create_configured_session, get_idp_url
 from ..webserver import AuthenticationWebRequestHandler, AuthenticationWebServer
 from .main import main
 from .tokens import create, request_api_key
@@ -20,7 +21,12 @@ def _perform_saml_authentication(
     opts, owner, enable_token_creation=False, use_stderr=False
 ):
     """Perform SAML authentication via web browser and local web server."""
-    session = create_configured_session(opts)
+    session = create_session(
+        proxy=opts.api_proxy,
+        ssl_verify=opts.api_ssl_verify,
+        user_agent=opts.api_user_agent,
+        headers=opts.api_headers,
+    )
     api_host = opts.api_config.host
 
     idp_url = get_idp_url(api_host, owner, session=session)

cloudsmith_cli/cli/config.py

@@ -152,21 +152,27 @@ def has_default_file(cls):
     @classmethod
     def load_config(cls, opts, path=None, profile=None):
         """Load a configuration file into an options object."""
-        if path and os.path.exists(path):
-            if os.path.isdir(path):
-                cls.config_searchpath.insert(0, path)
-            else:
-                cls.config_files.insert(0, path)
-
-        config = cls.read_config()
-        values = config.get("default", {})
-        cls._load_values_into_opts(opts, values)
-
-        if profile and profile != "default":
-            values = config.get("profile:%s" % profile, {})
+        orig_searchpath = cls.config_searchpath[:]
+        orig_files = cls.config_files[:]
+        try:
+            if path and os.path.exists(path):
+                if os.path.isdir(path):
+                    cls.config_searchpath.insert(0, path)
+                else:
+                    cls.config_files.insert(0, path)
+
+            config = cls.read_config()
+            values = config.get("default", {})
             cls._load_values_into_opts(opts, values)
 
-        return values
+            if profile and profile != "default":
+                values = config.get("profile:%s" % profile, {})
+                cls._load_values_into_opts(opts, values)
+
+            return values
+        finally:
+            cls.config_searchpath = orig_searchpath
+            cls.config_files = orig_files
 
     @staticmethod
     def _load_values_into_opts(opts, values):

cloudsmith_cli/cli/decorators.py

@@ -7,6 +7,8 @@
 from cloudsmith_cli.cli import validators
 
 from ..core.api.init import initialise_api as _initialise_api
+from ..core.config_resolver import ConfigResolver
+from ..core.credentials import CredentialContext, CredentialProviderChain
 from ..core.mcp import server
 from . import config, utils
 
@@ -221,8 +223,17 @@ def wrapper(ctx, *args, **kwargs):
     return wrapper
 
 
-def initialise_api(f):
-    """Initialise the Cloudsmith API for use."""
+def _pop_boolean_option(kwargs, name, invert=False):
+    """Pop a boolean option from kwargs, optionally inverting it."""
+    value = kwargs.pop(name)
+    value = value if value is not None else None
+    if value is not None and invert:
+        value = not value
+    return value
+
+
+def resolve_credentials(f):
+    """Resolve client configuration and credentials without initialising the API."""
 
     @click.option(
         "--api-host", envvar="CLOUDSMITH_API_HOST", help="The API host to connect to."
@@ -252,6 +263,75 @@ def initialise_api(f):
         envvar="CLOUDSMITH_API_HEADERS",
         help="A CSV list of extra headers (key=value) to send to the API.",
     )
+    @click.pass_context
+    @functools.wraps(f)
+    def wrapper(ctx, *args, **kwargs):
+        # pylint: disable=missing-docstring
+        opts = config.get_or_create_options(ctx)
+
+        cli_api_host = kwargs.pop("api_host")
+        cli_api_proxy = kwargs.pop("api_proxy")
+        cli_ssl_verify = _pop_boolean_option(
+            kwargs, "without_api_ssl_verify", invert=True
+        )
+        cli_api_user_agent = kwargs.pop("api_user_agent")
+        cli_api_headers = kwargs.pop("api_headers")
+
+        client_config = ConfigResolver().resolve(
+            api_host=cli_api_host,
+            proxy=cli_api_proxy,
+            ssl_verify=cli_ssl_verify,
+            user_agent=cli_api_user_agent,
+            headers=cli_api_headers,
+            debug=opts.debug,
+            profile=ctx.meta.get("profile"),
+            config_file=ctx.meta.get("config_file"),
+        )
+
+        opts.api_host = client_config.api_host
+        opts.api_proxy = client_config.proxy
+        opts.api_ssl_verify = client_config.ssl_verify
+        opts.api_user_agent = client_config.user_agent
+        opts.api_headers = client_config.headers
+
+        credential_context = CredentialContext(
+            config=client_config,
+            creds_file_path=ctx.meta.get("creds_file"),
+            profile=ctx.meta.get("profile"),
+            debug=opts.debug,
+            cli_api_key=opts.api_key,
+        )
+
+        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,
+                )
+
+        opts.client_config = client_config
+        opts.credential = credential
+
+        kwargs["opts"] = opts
+        return ctx.invoke(f, *args, **kwargs)
+
+    return wrapper
+
+
+def initialise_api(f):
+    """Initialise the Cloudsmith API for use. Depends on resolve_credentials."""
+
     @click.option(
         "-R",
         "--without-rate-limit",
@@ -294,49 +374,50 @@ def initialise_api(f):
     @functools.wraps(f)
     def wrapper(ctx, *args, **kwargs):
         # pylint: disable=missing-docstring
-        def _set_boolean(name, invert=False):
-            value = kwargs.pop(name)
-            value = value if value is not None else None
-            if value is not None and invert:
-                value = not value
-            return value
-
         opts = config.get_or_create_options(ctx)
-        opts.api_host = kwargs.pop("api_host")
-        opts.api_proxy = kwargs.pop("api_proxy")
-        opts.api_ssl_verify = _set_boolean("without_api_ssl_verify", invert=True)
-        opts.api_user_agent = kwargs.pop("api_user_agent")
-        opts.api_headers = kwargs.pop("api_headers")
-        opts.rate_limit = _set_boolean("without_rate_limit", invert=True)
+
+        opts.rate_limit = _pop_boolean_option(kwargs, "without_rate_limit", invert=True)
         opts.rate_limit_warning = kwargs.pop("rate_limit_warning")
         opts.error_retry_max = kwargs.pop("error_retry_max")
         opts.error_retry_backoff = kwargs.pop("error_retry_backoff")
         opts.error_retry_codes = kwargs.pop("error_retry_codes")
         opts.error_retry_cb = report_retry
 
+        client_config = opts.client_config
+        credential = opts.credential
+
+        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,
-            proxy=opts.api_proxy,
-            ssl_verify=opts.api_ssl_verify,
-            user_agent=opts.api_user_agent,
-            headers=opts.api_headers,
+            debug=client_config.debug,
+            host=client_config.api_host,
+            key=resolved_key,
+            proxy=client_config.proxy,
+            ssl_verify=client_config.ssl_verify,
+            user_agent=client_config.user_agent,
+            headers=client_config.headers,
             rate_limit=opts.rate_limit,
             rate_limit_callback=call_print_rate_limit_info_with_opts,
             error_retry_max=opts.error_retry_max,
             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
         return ctx.invoke(f, *args, **kwargs)
 
-    return wrapper
+    return resolve_credentials(wrapper)
 
 
 def initialise_mcp(f):

cloudsmith_cli/cli/saml.py

@@ -11,9 +11,7 @@
 @pytest.fixture
 def mock_saml_session():
     """Mock the SAML session creation."""
-    with patch(
-        "cloudsmith_cli.cli.commands.auth.create_configured_session"
-    ) as mock_session:
+    with patch("cloudsmith_cli.cli.commands.auth.create_session") as mock_session:
         mock_session.return_value = MagicMock()
         yield mock_session
 

cloudsmith_cli/cli/tests/commands/test_auth.py

@@ -4,7 +4,7 @@
 import requests
 
 from ...core.api.exceptions import ApiException
-from ..saml import exchange_2fa_token, get_idp_url, refresh_access_token
+from ...core.saml import exchange_2fa_token, get_idp_url, refresh_access_token
 
 
 @pytest.fixture

cloudsmith_cli/cli/tests/test_saml.py

@@ -10,7 +10,7 @@
 from ..core.api.exceptions import ApiException
 from ..core.api.init import initialise_api
 from ..core.keyring import store_sso_tokens
-from .saml import exchange_2fa_token
+from ..core.saml import exchange_2fa_token
 
 
 def get_template_path(template_name):