feat: add OIDC credential auto-discovery

.pylintrc

@@ -455,7 +455,9 @@ disable=raw-checker-failed,
         used-before-assignment,
         unneeded-not,
         duplicate-code,
-        cyclic-import
+        cyclic-import,
+        too-many-public-methods,
+        too-many-instance-attributes
 
 
 # Enable the message, report, category or checker with the given id(s). You can

README.md

@@ -136,6 +136,30 @@ Or you can get the latest pre-release version from Cloudsmith:
 pip install --upgrade cloudsmith-cli --extra-index-url=https://dl.cloudsmith.io/public/cloudsmith/cli/python/index/
 ```
 
+### Optional Dependencies
+
+The CLI supports optional extras for additional functionality:
+
+#### AWS OIDC Support
+
+For AWS environments (ECS, EKS, EC2), install with `aws` extra to enable automatic credential discovery:
+
+```
+pip install cloudsmith-cli[aws]
+```
+
+This installs `boto3[crt]` for AWS credential chain support, STS token generation, and AWS SSO compatibility.
+
+#### All Optional Features
+
+To install all optional dependencies:
+
+```
+pip install cloudsmith-cli[all]
+```
+
+**Note:** If you don't install the AWS extra, the AWS OIDC detector will gracefully skip itself with no errors.
+
 ## Configuration
 
 There are two configuration files used by the CLI:

cloudsmith_cli/cli/commands/whoami.py

@@ -1,14 +1,11 @@
 """CLI/Commands - Retrieve authentication status."""
 
-import os
-
 import click
 
 from ...core import keyring
 from ...core.api.exceptions import ApiException
 from ...core.api.user import get_token_metadata, get_user_brief
 from .. import decorators, utils
-from ..config import CredentialsReader
 from ..exceptions import handle_api_exceptions
 from .main import main
 
@@ -26,26 +23,17 @@ def _get_active_method(api_config):
 def _get_api_key_source(opts):
     """Determine where the API key was loaded from.
 
-    Checks in priority order matching actual resolution:
-    CLI --api-key flag > CLOUDSMITH_API_KEY env var > credentials.ini.
+    Uses the credential provider chain result attached by initialise_api.
     """
-    if not opts.api_key:
-        return {"configured": False, "source": None, "source_key": None}
-
-    env_key = os.environ.get("CLOUDSMITH_API_KEY")
-
-    # If env var is set but differs from the resolved key, CLI flag won
-    if env_key and opts.api_key != env_key:
-        source, key = "CLI --api-key flag", "cli_flag"
-    elif env_key:
-        suffix = env_key[-4:]
-        source, key = f"CLOUDSMITH_API_KEY env var (ends with ...{suffix})", "env_var"
-    elif creds := CredentialsReader.find_existing_files():
-        source, key = f"credentials.ini ({creds[0]})", "credentials_file"
-    else:
-        source, key = "CLI --api-key flag", "cli_flag"
+    credential = getattr(opts, "credential", None)
+    if credential:
+        return {
+            "configured": True,
+            "source": credential.source_detail or credential.source_name,
+            "source_key": credential.source_name,
+        }
 
-    return {"configured": True, "source": source, "source_key": key}
+    return {"configured": False, "source": None, "source_key": None}
 
 
 def _get_sso_status(api_host):
@@ -120,7 +108,12 @@ def _print_verbose_text(data):
                 click.echo(f"  Source: {ak['source']}")
             click.echo("  Note: SSO token is being used instead")
     elif active == "api_key":
-        click.secho("Authentication Method: API Key", fg="cyan", bold=True)
+        if ak.get("source_key") == "oidc":
+            click.secho(
+                "Authentication Method: OIDC Auto-Discovery", fg="cyan", bold=True
+            )
+        else:
+            click.secho("Authentication Method: API Key", fg="cyan", bold=True)
         for label, field in [
             ("Source", "source"),
             ("Token Slug", "slug"),

cloudsmith_cli/cli/config.py

@@ -66,6 +66,9 @@ class Default(SectionSchema):
         api_user_agent = ConfigParam(name="api_user_agent", type=str)
         mcp_allowed_tools = ConfigParam(name="mcp_allowed_tools", type=str)
         mcp_allowed_tool_groups = ConfigParam(name="mcp_allowed_tool_groups", type=str)
+        oidc_audience = ConfigParam(name="oidc_audience", type=str)
+        oidc_org = ConfigParam(name="oidc_org", type=str)
+        oidc_service_slug = ConfigParam(name="oidc_service_slug", type=str)
 
     @matches_section("profile:*")
     class Profile(Default):
@@ -416,6 +419,48 @@ def mcp_allowed_tool_groups(self, value):
 
         self._set_option("mcp_allowed_tool_groups", tool_groups)
 
+    @property
+    def oidc_audience(self):
+        """Get value for OIDC audience."""
+        return self._get_option("oidc_audience")
+
+    @oidc_audience.setter
+    def oidc_audience(self, value):
+        """Set value for OIDC audience."""
+        self._set_option("oidc_audience", value)
+
+    @property
+    def oidc_org(self):
+        """Get value for OIDC organisation slug."""
+        return self._get_option("oidc_org")
+
+    @oidc_org.setter
+    def oidc_org(self, value):
+        """Set value for OIDC organisation slug."""
+        self._set_option("oidc_org", value)
+
+    @property
+    def oidc_service_slug(self):
+        """Get value for OIDC service slug."""
+        return self._get_option("oidc_service_slug")
+
+    @oidc_service_slug.setter
+    def oidc_service_slug(self, value):
+        """Set value for OIDC service slug."""
+        self._set_option("oidc_service_slug", value)
+
+    @property
+    def oidc_discovery_disabled(self):
+        """Get value for OIDC discovery disabled flag."""
+        return self._get_option("oidc_discovery_disabled", default=False)
+
+    @oidc_discovery_disabled.setter
+    def oidc_discovery_disabled(self, value):
+        """Set value for OIDC discovery disabled flag."""
+        self._set_option(
+            "oidc_discovery_disabled", bool(value) if value is not None else False
+        )
+
     @property
     def output(self):
         """Get value for output format."""

cloudsmith_cli/cli/decorators.py

@@ -7,7 +7,10 @@
 from cloudsmith_cli.cli import validators
 
 from ..core.api.init import initialise_api as _initialise_api
+from ..core.credentials.chain import CredentialProviderChain
+from ..core.credentials.models import CredentialContext
 from ..core.mcp import server
+from ..core.rest import create_requests_session as _create_session
 from . import config, utils
 
 
@@ -20,6 +23,14 @@ def report_retry(seconds, context=None):
         )
 
 
+def _pop_boolean_flag(kwargs, name, invert=False):
+    """Pop a boolean flag from kwargs, optionally inverting it."""
+    value = kwargs.pop(name)
+    if value is not None and invert:
+        value = not value
+    return value
+
+
 def common_package_action_options(f):
     """Add common options for package actions."""
 
@@ -214,15 +225,17 @@ def common_api_auth_options(f):
     def wrapper(ctx, *args, **kwargs):
         # pylint: disable=missing-docstring
         opts = config.get_or_create_options(ctx)
-        opts.api_key = kwargs.pop("api_key")
+        api_key = kwargs.pop("api_key")
+        if api_key:
+            opts.api_key = api_key
         kwargs["opts"] = opts
         return ctx.invoke(f, *args, **kwargs)
 
     return wrapper
 
 
-def initialise_api(f):
-    """Initialise the Cloudsmith API for use."""
+def initialise_session(f):
+    """Create a shared HTTP session with proxy/SSL/user-agent settings."""
 
     @click.option(
         "--api-host", envvar="CLOUDSMITH_API_HOST", help="The API host to connect to."
@@ -252,6 +265,118 @@ 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)
+        opts.api_host = kwargs.pop("api_host")
+        opts.api_proxy = kwargs.pop("api_proxy")
+        opts.api_ssl_verify = _pop_boolean_flag(
+            kwargs, "without_api_ssl_verify", invert=True
+        )
+        opts.api_user_agent = kwargs.pop("api_user_agent")
+        opts.api_headers = kwargs.pop("api_headers")
+
+        opts.session = _create_session(
+            proxy=opts.api_proxy,
+            ssl_verify=opts.api_ssl_verify,
+            user_agent=opts.api_user_agent,
+            headers=opts.api_headers,
+        )
+
+        kwargs["opts"] = opts
+        return ctx.invoke(f, *args, **kwargs)
+
+    return wrapper
+
+
+def resolve_credentials(f):
+    """Resolve credentials via the provider chain. Depends on initialise_session."""
+
+    @click.option(
+        "--oidc-audience",
+        envvar="CLOUDSMITH_OIDC_AUDIENCE",
+        help="The OIDC audience for token requests.",
+    )
+    @click.option(
+        "--oidc-org",
+        envvar="CLOUDSMITH_ORG",
+        help="The Cloudsmith organisation slug for OIDC token exchange.",
+    )
+    @click.option(
+        "--oidc-service-slug",
+        envvar="CLOUDSMITH_SERVICE_SLUG",
+        help="The Cloudsmith service slug for OIDC token exchange.",
+    )
+    @click.option(
+        "--oidc-discovery-disabled",
+        default=None,
+        is_flag=True,
+        envvar="CLOUDSMITH_OIDC_DISCOVERY_DISABLED",
+        help="Disable OIDC auto-discovery.",
+    )
+    @click.pass_context
+    @functools.wraps(f)
+    def wrapper(ctx, *args, **kwargs):
+        # pylint: disable=missing-docstring
+        opts = config.get_or_create_options(ctx)
+
+        oidc_audience = kwargs.pop("oidc_audience")
+        oidc_org = kwargs.pop("oidc_org")
+        oidc_service_slug = kwargs.pop("oidc_service_slug")
+        oidc_discovery_disabled = _pop_boolean_flag(kwargs, "oidc_discovery_disabled")
+
+        if oidc_audience:
+            opts.oidc_audience = oidc_audience
+        if oidc_org:
+            opts.oidc_org = oidc_org
+        if oidc_service_slug:
+            opts.oidc_service_slug = oidc_service_slug
+        if oidc_discovery_disabled:
+            opts.oidc_discovery_disabled = oidc_discovery_disabled
+
+        context = CredentialContext(
+            session=opts.session,
+            api_key=opts.api_key,
+            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,
+            oidc_audience=opts.oidc_audience,
+            oidc_org=opts.oidc_org,
+            oidc_service_slug=opts.oidc_service_slug,
+            oidc_discovery_disabled=opts.oidc_discovery_disabled,
+        )
+
+        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 credential:
+                click.secho(
+                    "Falling back to API key authentication.",
+                    fg="yellow",
+                    err=True,
+                )
+
+        opts.credential = credential
+
+        kwargs["opts"] = opts
+        return ctx.invoke(f, *args, **kwargs)
+
+    return initialise_session(wrapper)
+
+
+def initialise_api(f):
+    """Initialise the Cloudsmith API for use. Depends on resolve_credentials."""
+
     @click.option(
         "-R",
         "--without-rate-limit",
@@ -294,20 +419,8 @@ 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_flag(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")
@@ -320,7 +433,7 @@ def call_print_rate_limit_info_with_opts(rate_info):
         opts.api_config = _initialise_api(
             debug=opts.debug,
             host=opts.api_host,
-            key=opts.api_key,
+            credential=opts.credential,
             proxy=opts.api_proxy,
             ssl_verify=opts.api_ssl_verify,
             user_agent=opts.api_user_agent,
@@ -336,7 +449,7 @@ def call_print_rate_limit_info_with_opts(rate_info):
         kwargs["opts"] = opts
         return ctx.invoke(f, *args, **kwargs)
 
-    return wrapper
+    return resolve_credentials(wrapper)
 
 
 def initialise_mcp(f):

cloudsmith_cli/cli/tests/conftest.py

@@ -5,6 +5,7 @@
 
 from ...core.api.init import initialise_api
 from ...core.api.repos import create_repo, delete_repo
+from ...core.credentials.models import CredentialResult
 from .utils import random_str
 
 
@@ -51,7 +52,9 @@ def organization():
 @pytest.fixture()
 def tmp_repository(organization, api_host, api_key):
     """Yield a temporary repository."""
-    initialise_api(host=api_host, key=api_key)
+    initialise_api(
+        host=api_host, credential=CredentialResult(api_key=api_key, source_name="test")
+    )
     repo_data = create_repo(organization, {"name": random_str()})
     yield repo_data
     delete_repo(organization, repo_data["slug"])