Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 2 additions & 0 deletions CHANGELOG.md
Original file line number Diff line number Diff line change
Expand Up @@ -15,6 +15,8 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
### Changed

- `cloudsmith domains list` now includes the Workspace slug in a `workspace` field.
- `cloudsmith auth` now reuses an existing SSO session when it can be renewed.
- `cloudsmith auth` now reports the SSO access-token expiry in normal and JSON output.

### Fixed

Expand Down
73 changes: 68 additions & 5 deletions cloudsmith_cli/cli/commands/auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -3,7 +3,13 @@
import webbrowser

import click
import requests

from ...core.sso import (
SsoRenewalStatus,
get_access_token_expiry,
renew_sso_session,
)
from .. import decorators, utils, validators
from ..exceptions import handle_api_exceptions
from ..saml import create_configured_session, get_idp_url
Expand Down Expand Up @@ -46,6 +52,55 @@ def _create_auth_server(opts, owner, session, enable_token_creation, profile):
) from last_error


def _renew_existing_sso_session(opts, profile):
"""Return a usable existing SSO session, or fall back to browser auth."""
renewal = renew_sso_session(opts.api_config.host, opts.session, profile=profile)
if renewal.status in (
SsoRenewalStatus.RENEWED,
SsoRenewalStatus.CURRENT,
):
return renewal
if renewal.status == SsoRenewalStatus.FAILED and isinstance(
renewal.error, requests.RequestException
):
raise click.ClickException(
"The SSO session has expired and Cloudsmith could not be reached. "
"Check your connection, then run 'cloudsmith auth' again."
) from renewal.error
return None


def _report_active_sso_session(opts, renewal, use_stderr=False):
"""Report the usable SSO session without exposing its access token."""
expires_at = get_access_token_expiry(renewal.access_token)
data = {
"authenticated": True,
"method": "sso",
"status": renewal.status.value,
"expires_at": expires_at,
"renewal_error": str(renewal.error) if renewal.error else None,
}
if utils.maybe_print_as_json(opts, data):
return

if renewal.status == SsoRenewalStatus.RENEWED:
click.secho("SSO session renewed.", fg="green", err=use_stderr)
elif renewal.error:
click.secho(
"The SSO session could not be renewed; the existing access token "
"remains active.",
fg="yellow",
err=use_stderr,
)
else:
click.secho("SSO session is active.", fg="green", err=use_stderr)
if expires_at:
click.echo(
f"Access token expires at {utils.fmt_datetime(expires_at)}.",
err=use_stderr,
)


def _perform_saml_authentication(
opts,
owner,
Expand All @@ -64,6 +119,11 @@ def _perform_saml_authentication(
redirect_url = f"http://{AUTH_REDIRECT_HOST}:{port}"

try:
click.echo(
f"Waiting for the authentication callback on port {port} ... ",
err=use_stderr,
)

idp_url = get_idp_url(
api_host, owner, redirect_url=redirect_url, session=session
)
Expand Down Expand Up @@ -96,11 +156,6 @@ def _perform_saml_authentication(
err=use_stderr,
)

click.echo(
f"Waiting for the authentication callback on port {port} ... ",
err=use_stderr,
)

auth_server.handle_request()
finally:
auth_server.server_close()
Expand Down Expand Up @@ -197,6 +252,14 @@ def authenticate(
err=True,
)

if not force and not token and not request_api_key_flag:
session_result = _renew_existing_sso_session(
opts, profile=ctx.meta.get("profile")
)
if session_result:
Comment thread
cloudsmith-iduffy marked this conversation as resolved.
_report_active_sso_session(opts, session_result, use_stderr)
return

workspace = opts.org or click.prompt("Workspace", err=use_stderr)
workspace = validators.validate_owner(ctx, None, workspace)[0]
opts.org = workspace
Expand Down
139 changes: 138 additions & 1 deletion cloudsmith_cli/cli/tests/commands/test_auth.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,16 +2,32 @@

import json
import webbrowser
from datetime import datetime, timezone
from unittest.mock import ANY, MagicMock, patch

import jwt
import pytest
import requests

from ....core.api.exceptions import ApiException
from ...commands.auth import authenticate
from ....core.sso import SsoRenewalResult, SsoRenewalStatus
from ...commands.auth import _renew_existing_sso_session, authenticate
from ...commands.main import main
from .conftest import MockToken


@pytest.fixture(autouse=True)
def no_existing_sso_session(monkeypatch):
"""Keep browser-flow tests independent of locally stored SSO sessions."""
monkeypatch.delenv("CLOUDSMITH_WORKSPACE", raising=False)
monkeypatch.delenv("CLOUDSMITH_ORG", raising=False)
with patch(
"cloudsmith_cli.cli.commands.auth._renew_existing_sso_session",
return_value=None,
):
yield


@pytest.fixture
def mock_saml_session():
"""Mock the SAML session creation."""
Expand Down Expand Up @@ -186,6 +202,30 @@ def test_auth_command_fails_after_all_redirect_ports_are_unavailable(
assert isinstance(result.exception, SystemExit)
assert "12400, 12401, 12402, 12403, 12404" in result.output

def test_auth_command_reports_callback_port_before_idp_lookup(
self,
runner,
mock_saml_session,
mock_webbrowser,
mock_auth_server,
):
"""Verify the callback port is shown even if IDP URL lookup fails."""
with patch(
"cloudsmith_cli.cli.commands.auth.get_idp_url",
side_effect=RuntimeError("lookup failed"),
):
result = runner.invoke(
authenticate,
["--owner", "testorg"],
catch_exceptions=True,
)

assert result.exit_code != 0
assert "Waiting for the authentication callback on port 12400" in result.output
mock_webbrowser.open.assert_not_called()
mock_auth_server.return_value.handle_request.assert_not_called()
mock_auth_server.return_value.server_close.assert_called_once()

def test_auth_command_opens_browser(
self,
runner,
Expand Down Expand Up @@ -228,6 +268,103 @@ def test_auth_command_passes_owner_to_webserver(
call_kwargs = mock_auth_server.call_args.kwargs
assert call_kwargs.get("owner") == "testorg"

def test_usable_session_avoids_workspace_and_browser(self, runner):
expires_at = datetime(2030, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
access_token = jwt.encode(
{"exp": expires_at},
"not-used-for-verification",
algorithm="HS256",
)
renewal = SsoRenewalResult(
status=SsoRenewalStatus.RENEWED,
access_token=access_token,
)
with (
patch(
"cloudsmith_cli.cli.commands.auth._renew_existing_sso_session",
return_value=renewal,
),
patch("cloudsmith_cli.cli.commands.auth.webbrowser") as browser,
patch(
"cloudsmith_cli.cli.commands.auth.AuthenticationWebServer"
) as auth_server,
):
result = runner.invoke(authenticate, [], catch_exceptions=False)

assert result.exit_code == 0
assert result.stdout == (
"SSO session renewed.\nAccess token expires at 2030-01-02T03:04:05Z.\n"
)
assert "Workspace" not in result.output
browser.open.assert_not_called()
auth_server.assert_not_called()

@pytest.mark.parametrize("output_format", ["json", "pretty_json"])
def test_json_renewal_report_is_machine_readable(self, runner, output_format):
expires_at = datetime(2030, 1, 2, 3, 4, 5, tzinfo=timezone.utc)
renewal = SsoRenewalResult(
status=SsoRenewalStatus.CURRENT,
access_token=jwt.encode(
{"exp": expires_at},
"not-used-for-verification",
algorithm="HS256",
),
)
with patch(
"cloudsmith_cli.cli.commands.auth._renew_existing_sso_session",
return_value=renewal,
):
result = runner.invoke(
authenticate,
["--output-format", output_format],
catch_exceptions=False,
)
legacy_result = (
runner.invoke(authenticate, ["--json"], catch_exceptions=False)
if output_format == "json"
else None
)

assert result.exit_code == 0
assert json.loads(result.stdout)["data"] == {
"authenticated": True,
"expires_at": "2030-01-02T03:04:05Z",
"method": "sso",
"renewal_error": None,
"status": "current",
}
assert result.stderr == ""
if legacy_result:
assert legacy_result.stdout == ""
assert "Access token expires at 2030-01-02T03:04:05Z." in (
legacy_result.stderr
)

def test_expired_session_offline_has_actionable_error(self, runner):
renewal = SsoRenewalResult(
status=SsoRenewalStatus.FAILED,
error=requests.ConnectionError("offline"),
)
with (
patch(
"cloudsmith_cli.cli.commands.auth.renew_sso_session",
return_value=renewal,
),
patch(
"cloudsmith_cli.cli.commands.auth._renew_existing_sso_session",
wraps=_renew_existing_sso_session,
),
patch(
"cloudsmith_cli.cli.commands.auth.AuthenticationWebServer"
) as auth_server,
):
result = runner.invoke(authenticate, [])

assert result.exit_code == 1
assert "Cloudsmith could not be reached" in result.output
assert "Check your connection" in result.output
auth_server.assert_not_called()


class TestBrowserFallback:
"""Tests for graceful handling of webbrowser.open() failures."""
Expand Down
Loading