feat: add HTML templates for authentication success and error, and improve session handling in SSO

Author: cloudsmith-iduffy

.envrc

@@ -18,10 +18,11 @@ if [ -d ".venv" ]; then
   fi
 fi
 
-pyenv install --skip-existing "$PYTHON_VERSION"
-layout pyenv "$PYTHON_VERSION"
-python -m pip install -U pip
-python -m pip install -q -r requirements.txt
+uv venv .venv --python $PYTHON_VERSION
+uv pip install -U pip uv
+uv pip install -e .
+
+source ./.venv/bin/activate
 
 # Load environment variables from the .env file (for pytest, mainly)
 if [ ! -f .env ]; then

cloudsmith_cli/__init__.py

@@ -1,6 +1,9 @@
 """Cloudsmith CLI."""
+import warnings
+
 import click
 import urllib3
 
 click.disable_unicode_literals_warning = True
 urllib3.disable_warnings(urllib3.exceptions.InsecureRequestWarning)
+warnings.filterwarnings("ignore", category=ResourceWarning)

cloudsmith_cli/cli/commands/auth.py

@@ -5,7 +5,7 @@
 
 from .. import decorators, validators
 from ..exceptions import handle_api_exceptions
-from ..saml import get_idp_url
+from ..saml import create_configured_session, get_idp_url
 from ..webserver import AuthenticationWebRequestHandler, AuthenticationWebServer
 from .main import main
 
@@ -35,9 +35,11 @@ def authenticate(ctx, opts, owner):
         )
     )
 
+    session = create_configured_session(opts)
+
     context_message = "Failed to authenticate via SSO!"
     with handle_api_exceptions(ctx, opts=opts, context_msg=context_message):
-        idp_url = get_idp_url(api_host, owner)
+        idp_url = get_idp_url(api_host, owner, session=session)
         click.echo(
             "Opening your organization's SAML IDP URL in your browser: %(idp_url)s"
             % {"idp_url": click.style(idp_url, bold=True)}
@@ -51,8 +53,7 @@ def authenticate(ctx, opts, owner):
             AuthenticationWebRequestHandler,
             api_host=api_host,
             owner=owner,
+            session=session,
+            debug=opts.debug,
         )
         auth_server.handle_request()
-
-        click.echo()
-        click.secho("Authentication complete", fg="green")

cloudsmith_cli/cli/commands/whoami.py

@@ -5,7 +5,7 @@
 from ...core.api.user import get_user_brief
 from .. import decorators
 from ..exceptions import handle_api_exceptions
-from ..utils import maybe_spinner
+from ..utils import maybe_print_as_json, maybe_spinner
 from .main import main
 
 
@@ -17,13 +17,30 @@
 @click.pass_context
 def whoami(ctx, opts):
     """Retrieve your current authentication status."""
-    click.echo("Retrieving your authentication status from the API ... ", nl=False)
+    use_stderr = opts.output in ("json", "pretty_json")
+
+    click.echo(
+        "Retrieving your authentication status from the API ... ",
+        nl=False,
+        err=use_stderr,
+    )
 
     context_msg = "Failed to retrieve your authentication status!"
     with handle_api_exceptions(ctx, opts=opts, context_msg=context_msg):
         with maybe_spinner(opts):
             is_auth, username, email, name = get_user_brief()
-    click.secho("OK", fg="green")
+    click.secho("OK", fg="green", err=use_stderr)
+
+    data = {
+        "is_authenticated": is_auth,
+        "username": username,
+        "email": email,
+        "name": name,
+    }
+
+    if maybe_print_as_json(opts, data):
+        return
+
     click.echo("You are authenticated as:")
     if not is_auth:
         click.secho("Nobody (i.e. anonymous user)", fg="yellow")

cloudsmith_cli/cli/saml.py

@@ -5,14 +5,35 @@
 from ..core.api.exceptions import ApiException
 
 
-def get_idp_url(api_host, owner):
+def create_configured_session(opts):
+    """
+    Create a requests session configured with the options from opts.
+    """
+    session = requests.Session()
+
+    if hasattr(opts, "api_ssl_verify") and opts.api_ssl_verify is not None:
+        session.verify = opts.api_ssl_verify
+
+    if hasattr(opts, "api_proxy") and opts.api_proxy:
+        session.proxies = {"http": opts.api_proxy, "https": opts.api_proxy}
+
+    if hasattr(opts, "api_user_agent") and opts.api_user_agent:
+        session.headers.update({"User-Agent": opts.api_user_agent})
+
+    if hasattr(opts, "api_headers") and opts.api_headers:
+        session.headers.update(opts.api_headers)
+
+    return session
+
+
+def get_idp_url(api_host, owner, session):
     org_saml_url = "{api_host}/orgs/{owner}/saml/?{params}".format(
         api_host=api_host,
         owner=owner,
         params=urlencode({"redirect_url": "http://localhost:12400"}),
     )
 
-    org_saml_response = requests.get(org_saml_url, timeout=30)
+    org_saml_response = session.get(org_saml_url, timeout=30)
 
     try:
         org_saml_response.raise_for_status()
@@ -26,18 +47,20 @@ def get_idp_url(api_host, owner):
     return org_saml_response.json().get("redirect_url")
 
 
-def exchange_2fa_token(api_host, two_factor_token, totp_token):
+def exchange_2fa_token(api_host, two_factor_token, totp_token, session):
     exchange_data = {"two_factor_token": two_factor_token, "totp_token": totp_token}
     exchange_url = "{api_host}/user/two-factor/".format(api_host=api_host)
 
-    exchange_response = requests.post(
+    headers = {
+        "Authorization": "Bearer {two_factor_token}".format(
+            two_factor_token=two_factor_token
+        )
+    }
+
+    exchange_response = session.post(
         exchange_url,
         data=exchange_data,
-        headers={
-            "Authorization": "Bearer {two_factor_token}".format(
-                two_factor_token=two_factor_token
-            )
-        },
+        headers=headers,
         timeout=30,
     )
 
@@ -57,16 +80,18 @@ def exchange_2fa_token(api_host, two_factor_token, totp_token):
     return (access_token, refresh_token)
 
 
-def refresh_access_token(api_host, access_token, refresh_token):
+def refresh_access_token(api_host, access_token, refresh_token, session):
     data = {"refresh_token": refresh_token}
     url = "{api_host}/user/refresh-token/".format(api_host=api_host)
 
-    response = requests.post(
+    headers = {
+        "Authorization": "Bearer {access_token}".format(access_token=access_token)
+    }
+
+    response = session.post(
         url,
         data=data,
-        headers={
-            "Authorization": "Bearer {access_token}".format(access_token=access_token)
-        },
+        headers=headers,
         timeout=30,
     )
 

cloudsmith_cli/cli/tests/test_saml.py

@@ -35,7 +35,10 @@ def test_get_idp_url(self, mock_get_request, mock_response):
         mock_get_request.return_value = mock_response
         mock_response.json.return_value = {"redirect_url": "response_redirect_url"}
 
-        assert get_idp_url(self.api_host, "test_org") == "response_redirect_url"
+        assert (
+            get_idp_url(self.api_host, "test_org", session=requests.sessions.Session())
+            == "response_redirect_url"
+        )
         mock_get_request.assert_called_once_with(
             f"{self.api_host}/orgs/test_org/saml/?{self.query_params}", timeout=30
         )
@@ -50,7 +53,7 @@ def test_get_idp_url_with_request_error(self, mock_get_request, mock_response):
         )
 
         with pytest.raises(ApiException) as exc:
-            get_idp_url(self.api_host, "test_org")
+            get_idp_url(self.api_host, "test_org", session=requests.sessions.Session())
 
             assert exc == ApiException(
                 status=500, headers={"foo": "bar"}, body="Error body"
@@ -66,7 +69,12 @@ def test_exchange_2fa_token(self, mock_post_request, mock_response):
             "refresh_token": "refresh_token",
         }
 
-        assert exchange_2fa_token(self.api_host, "two_factor_token", "totp_token") == (
+        assert exchange_2fa_token(
+            self.api_host,
+            "two_factor_token",
+            "totp_token",
+            session=requests.sessions.Session(),
+        ) == (
             "access_token",
             "refresh_token",
         )
@@ -89,7 +97,12 @@ def test_exchange_2fa_token_with_request_error(
         )
 
         with pytest.raises(ApiException) as exc:
-            exchange_2fa_token(self.api_host, "two_factor_token", "totp_token")
+            exchange_2fa_token(
+                self.api_host,
+                "two_factor_token",
+                "totp_token",
+                session=requests.sessions.Session(),
+            )
 
             assert exc == ApiException(
                 status=500, headers={"foo": "bar"}, body="Error body"
@@ -111,7 +124,12 @@ def test_refresh_access_token(self, mock_post_request, mock_response):
             "refresh_token": "refresh_token",
         }
 
-        assert refresh_access_token(self.api_host, "access_token", "refresh_token") == (
+        assert refresh_access_token(
+            self.api_host,
+            "access_token",
+            "refresh_token",
+            session=requests.sessions.Session(),
+        ) == (
             "access_token",
             "refresh_token",
         )
@@ -138,7 +156,12 @@ def test_refresh_access_token_with_request_error(
         }
 
         with pytest.raises(ApiException) as exc:
-            exchange_2fa_token(self.api_host, "two_factor_token", "totp_token")
+            exchange_2fa_token(
+                self.api_host,
+                "two_factor_token",
+                "totp_token",
+                session=requests.sessions.Session(),
+            )
 
             assert exc == ApiException(
                 status=500, headers={"foo": "bar"}, body="Error body"