fix(ENG-13605): authenticate multi-part uploads via the credential chain

CHANGELOG.md

@@ -7,6 +7,12 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 
 ## [Unreleased]
 
+### Fixed
+
+- Files larger than 100MB now upload correctly when authenticated via SSO or OIDC. The multi-part upload read its credentials from `opts.api_key`, which is only populated by `--api-key`, `CLOUDSMITH_API_KEY` or `credentials.ini` — so with an SSO session or OIDC auto-discovery it was empty, the auth header was dropped, and the part upload failed with a misleading `404 - Not Found` ("this usually means the user/org is wrong or not visible") even though every preceding API call had succeeded. Credentials are now taken from the resolved credential chain, with SSO access tokens sent as a bearer `Authorization` header and API keys/OIDC tokens as `X-Api-Key`.
+- `cloudsmith download` now authenticates with OIDC credentials. It resolved its own auth from `opts.api_key` plus a direct keyring read, which between them cover neither OIDC auto-discovery nor any future credential source, so downloads from a private repository in an OIDC-authenticated pipeline were attempted anonymously. It now uses the same resolved credential as every other command.
+- The hint shown on a `401 - Unauthorized` now reflects how the session actually authenticated. It branched on `opts.api_key`, so an OIDC-authenticated session was told "You don't have an API key or access token set" despite being authenticated, and an SSO session whose token had expired could be told to check its permissions instead of to re-run `cloudsmith auth`.
+
 ## [1.20.1] - 2026-07-30
 
 ### Fixed

cloudsmith_cli/cli/decorators.py

@@ -246,9 +246,6 @@ def wrapper(ctx, *args, **kwargs):
             opts.api_key_from_flag = None
             opts.api_key_from_env = None
 
-        # Keep opts.api_key populated for any code that still reads it directly.
-        if api_key:
-            opts.api_key = api_key
         kwargs["opts"] = opts
         return ctx.invoke(f, *args, **kwargs)
 

cloudsmith_cli/cli/exceptions.py

@@ -7,7 +7,6 @@
 import click
 
 from ..core.api.exceptions import ApiException
-from ..core.keyring import get_access_token
 
 
 @contextlib.contextmanager
@@ -171,16 +170,17 @@ def get_error_hint(ctx, opts, exc):
 def get_401_error_hint(ctx, opts, exc):
     """Get the hint for a 401/Unauthorised error."""
     # pylint: disable=unused-argument
-    if opts.api_key:
+    credential = getattr(opts, "credential", None)
+
+    if credential and credential.auth_type == "bearer":
+        return "Since you have an SSO access token set, this probably means that it has expired. Try getting a new token with 'cloudsmith auth', then try again."
+
+    if credential:
         return (
             "Since you have an API key set, this probably means "
             "you don't have the permission to perform this action."
         )
 
-    access_token = get_access_token(opts.api_host)
-    if access_token:
-        return "Since you have an SSO access token set, this probably means that it has expired. Try getting a new token with 'cloudsmith auth', then try again."
-
     if ctx.info_name == "token":
         # This is already the token command
         return (

cloudsmith_cli/cli/tests/test_exceptions.py

@@ -0,0 +1,42 @@
+"""Tests for CLI error hints."""
+
+from unittest.mock import Mock
+
+from cloudsmith_cli.cli.exceptions import get_401_error_hint
+from cloudsmith_cli.core.credentials.models import CredentialResult
+
+
+def hint_for(credential, info_name="push"):
+    """Return the 401 hint for a session holding the given credential.
+
+    opts.api_key is left as an auto-Mock, so a hint that reads it rather than
+    the resolved credential fails these tests.
+    """
+    return get_401_error_hint(
+        Mock(info_name=info_name), Mock(credential=credential), Mock()
+    )
+
+
+class TestGet401ErrorHint:
+    """The hint has to describe the credential the chain actually resolved.
+    Branching on opts.api_key told OIDC sessions -- which never populate it --
+    that they had no credentials at all.
+    """
+
+    def test_bearer_credential_suggests_reauthenticating(self):
+        credential = CredentialResult(
+            api_key="sso-token", source_name="keyring", auth_type="bearer"
+        )
+
+        assert "cloudsmith auth" in hint_for(credential)
+
+    def test_api_key_credential_suggests_a_permissions_problem(self):
+        credential = CredentialResult(api_key="csa_abc123", source_name="oidc")
+
+        assert "permission" in hint_for(credential)
+
+    def test_no_credential_suggests_authenticating(self):
+        assert "cloudsmith token" in hint_for(None)
+
+    def test_no_credential_on_token_command_reports_a_failed_login(self):
+        assert "login failed" in hint_for(None, info_name="token")

cloudsmith_cli/core/api/files.py

@@ -97,10 +97,11 @@ def upload_file(upload_url, upload_fields, filepath, callback=None):
 def multi_part_upload_file(
     opts, upload_url, owner, repo, filepath, callback, upload_id
 ):
+    headers = opts.credential.auth_headers() if opts.credential else {}
+
     with open(filepath, "rb") as f:
         chunk_number = 1
         session = create_requests_session()
-        headers = {"X-Api-Key": opts.api_key}
         while chunk := f.read(CHUNK_SIZE):
             resp = session.put(
                 upload_url,

cloudsmith_cli/core/config.py

@@ -117,9 +117,9 @@ def new_config_messaging(has_errors, opts, create, api_key):
         click.secho("Oops, please fix the errors and try again!", fg="red")
         return
 
-    if opts.api_key != api_key:
+    if opts.api_key_from_file != api_key:
         click.echo()
-        if opts.api_key:
+        if opts.api_key_from_file:
             click.secho(
                 "Note: The above API key doesn't match what you have in "
                 "your default credentials config file.",

cloudsmith_cli/core/credentials/models.py

@@ -42,3 +42,15 @@ class CredentialResult:
     source_name: str
     source_detail: str | None = None
     auth_type: Literal["api_key", "bearer"] = "api_key"
+
+    def auth_headers(self) -> dict[str, str]:
+        """Return the headers that authenticate a raw request with this credential.
+
+        The two headers are not interchangeable: a raw JWT in ``X-Api-Key`` is
+        rejected outright, aborting the auth chain before the SSO
+        authenticators run.
+        """
+        if self.auth_type == "bearer":
+            return {"Authorization": f"Bearer {self.api_key}"}
+
+        return {"X-Api-Key": self.api_key}

cloudsmith_cli/core/download.py

@@ -5,12 +5,11 @@
 import os
 
 import click
-import cloudsmith_api
 import requests
 from rich.console import Console
 from rich.table import Table
 
-from . import keyring, ratelimits, utils
+from . import ratelimits, utils
 from .api.exceptions import catch_raise_api_exception
 from .api.packages import get_packages_api, list_packages
 from .rest import create_requests_session
@@ -36,21 +35,14 @@ def resolve_auth(
     headers = {}
     auth_source = "none"
 
-    # Follow the same authentication logic as the API initialization
-    # Priority: explicit --api-key > SSO token > configured API key
+    credential = getattr(opts, "credential", None)
 
-    # Only attempt keyring operations if keyring is enabled
-    config = cloudsmith_api.Configuration()
-    access_token = keyring.get_access_token(config.host)
-    api_key = api_key_opt or getattr(opts, "api_key", None)
-
-    if api_key:
-        # Prioritize API key (from --api-key option or CLOUDSMITH_API_KEY env var) over SSO
-        headers["X-Api-Key"] = api_key
+    if api_key_opt:
+        headers["X-Api-Key"] = api_key_opt
         auth_source = "api-key"
-    elif access_token:
-        headers["Authorization"] = f"Bearer {access_token}"
-        auth_source = "sso"
+    elif credential:
+        headers = credential.auth_headers()
+        auth_source = "sso" if credential.auth_type == "bearer" else "api-key"
 
     return session, headers, auth_source
 

cloudsmith_cli/core/tests/test_config.py

@@ -0,0 +1,43 @@
+"""Tests for config file messaging."""
+
+from unittest.mock import Mock, patch
+
+from ..config import new_config_messaging
+
+
+def capture_messages(api_key_from_file, new_api_key, create=False):
+    """Return the messaging emitted after generating config files, as one string."""
+    opts = Mock(api_key_from_file=api_key_from_file)
+
+    with (
+        patch("cloudsmith_cli.core.config.click.secho") as secho,
+        patch("cloudsmith_cli.core.config.click.echo"),
+    ):
+        new_config_messaging(
+            has_errors=False, opts=opts, create=create, api_key=new_api_key
+        )
+
+    return " ".join(str(call.args[0]) for call in secho.call_args_list if call.args)
+
+
+class TestNewConfigMessaging:
+    """The mismatch warning names the credentials file specifically, so it reads
+    the file-sourced chain input rather than the effective key.
+    """
+
+    def test_warns_when_the_file_key_differs_from_the_new_token(self):
+        messages = capture_messages(api_key_from_file="csa_old", new_api_key="csa_new")
+
+        assert "doesn't match" in messages
+
+    def test_stays_quiet_when_the_file_key_matches_the_new_token(self):
+        messages = capture_messages(
+            api_key_from_file="csa_same", new_api_key="csa_same"
+        )
+
+        assert "doesn't match" not in messages
+
+    def test_prompts_to_store_the_key_when_the_file_has_none(self):
+        messages = capture_messages(api_key_from_file=None, new_api_key="csa_new")
+
+        assert "Don't forget to put your API key in a config file" in messages

cloudsmith_cli/core/tests/test_download.py

@@ -9,6 +9,7 @@
 import click
 
 from cloudsmith_cli.core import download
+from cloudsmith_cli.core.credentials.models import CredentialResult
 
 
 class TestResolveAuth(unittest.TestCase):
@@ -19,27 +20,44 @@ def setUp(self):
         self.mock_opts.debug = False
         self.mock_opts.rate_limit = True
         self.mock_opts.error_retry_cb = None
-        self.mock_opts.api_key = None
+        self.mock_opts.credential = None
 
     @patch("cloudsmith_cli.core.download.create_requests_session")
-    def test_resolve_auth_api_key_from_opts(self, mock_create_session):
-        """Test auth resolution with API key from opts."""
+    def test_resolve_auth_api_key_credential(self, mock_create_session):
+        """Test auth resolution with an API key from the credential chain."""
         mock_session = Mock()
         mock_create_session.return_value = mock_session
-        self.mock_opts.api_key = "test-api-key"
+        self.mock_opts.credential = CredentialResult(
+            api_key="test-api-key", source_name="env_var"
+        )
 
         session, headers, auth_source = download.resolve_auth(self.mock_opts)
 
         self.assertEqual(session, mock_session)
         self.assertEqual(headers, {"X-Api-Key": "test-api-key"})
         self.assertEqual(auth_source, "api-key")
 
+    @patch("cloudsmith_cli.core.download.create_requests_session")
+    def test_resolve_auth_sso_credential(self, mock_create_session):
+        """Test that an SSO session resolved by the chain uses a bearer header."""
+        mock_create_session.return_value = Mock()
+        self.mock_opts.credential = CredentialResult(
+            api_key="sso-token", source_name="keyring", auth_type="bearer"
+        )
+
+        _session, headers, auth_source = download.resolve_auth(self.mock_opts)
+
+        self.assertEqual(headers, {"Authorization": "Bearer sso-token"})
+        self.assertEqual(auth_source, "sso")
+
     @patch("cloudsmith_cli.core.download.create_requests_session")
     def test_resolve_auth_api_key_override(self, mock_create_session):
         """Test auth resolution with API key override."""
         mock_session = Mock()
         mock_create_session.return_value = mock_session
-        self.mock_opts.api_key = "config-key"
+        self.mock_opts.credential = CredentialResult(
+            api_key="chain-key", source_name="keyring", auth_type="bearer"
+        )
 
         _session, headers, auth_source = download.resolve_auth(
             self.mock_opts, api_key_opt="override-key"
@@ -48,6 +66,16 @@ def test_resolve_auth_api_key_override(self, mock_create_session):
         self.assertEqual(headers, {"X-Api-Key": "override-key"})
         self.assertEqual(auth_source, "api-key")
 
+    @patch("cloudsmith_cli.core.download.create_requests_session")
+    def test_resolve_auth_no_credentials(self, mock_create_session):
+        """Test auth resolution when the chain resolved nothing."""
+        mock_create_session.return_value = Mock()
+
+        _session, headers, auth_source = download.resolve_auth(self.mock_opts)
+
+        self.assertEqual(headers, {})
+        self.assertEqual(auth_source, "none")
+
 
 class TestResolvePackage(unittest.TestCase):
     """Test package resolution logic."""

cloudsmith_cli/core/tests/test_files.py

@@ -0,0 +1,68 @@
+"""Tests for the files API upload helpers."""
+
+from unittest.mock import Mock, patch
+
+from ..api.files import multi_part_upload_file
+from ..credentials.models import CredentialResult
+
+
+def upload_and_capture_headers(tmp_path, credential, api_key=None):
+    """Run a single-chunk multi-part upload and return the headers it sent."""
+    filepath = tmp_path / "package.raw"
+    filepath.write_bytes(b"payload")
+
+    opts = Mock(credential=credential, api_key=api_key)
+    session = Mock()
+    session.put.return_value = Mock(raise_for_status=Mock(return_value=None))
+
+    with (
+        patch(
+            "cloudsmith_cli.core.api.files.create_requests_session",
+            return_value=session,
+        ),
+        patch("cloudsmith_cli.core.api.files.get_files_api"),
+    ):
+        multi_part_upload_file(
+            opts,
+            upload_url="https://upload.example.invalid/parts",
+            owner="owner",
+            repo="repo",
+            filepath=str(filepath),
+            callback=lambda: None,
+            upload_id="upload-id",
+        )
+
+    return session.put.call_args.kwargs["headers"]
+
+
+class TestMultiPartUploadAuth:
+    """The multi-part upload endpoint answers 404 for an unauthenticated
+    request, so credentials must come from the resolved credential chain
+    rather than opts.api_key, which OIDC and SSO never populate.
+    """
+
+    def test_uses_resolved_credential_not_opts_api_key(self, tmp_path):
+        credential = CredentialResult(api_key="csa_resolved", source_name="oidc")
+
+        headers = upload_and_capture_headers(tmp_path, credential, api_key="csa_stale")
+
+        assert headers == {"X-Api-Key": "csa_resolved"}
+
+    def test_invents_no_header_without_a_credential(self, tmp_path):
+        """No credential means no credentials at all, and the preceding
+        files_create call would already have failed.
+        """
+        headers = upload_and_capture_headers(
+            tmp_path, credential=None, api_key="csa_stale"
+        )
+
+        assert headers == {}
+
+    def test_sends_sso_token_as_bearer_authorization(self, tmp_path):
+        credential = CredentialResult(
+            api_key="sso-token", source_name="keyring", auth_type="bearer"
+        )
+
+        headers = upload_and_capture_headers(tmp_path, credential)
+
+        assert headers == {"Authorization": "Bearer sso-token"}