Fix SAML authentication when redirect port is occupied

README.md

@@ -321,7 +321,7 @@ cloudsmith auth --owner example
 Beginning authentication for the example org ...
 Your organization's SAML IDP URL is: https://example.com/some-saml-idp
 
-Starting webserver to begin authentication ...
+Waiting for the authentication callback on port 12400 ...
 
 Authentication complete
 ```
@@ -332,7 +332,7 @@ The CLI opens the IDP URL in your browser automatically. If it can't (for exampl
 cloudsmith auth --owner example --no-browser
 ```
 
-*Note:* The authentication callback is served on `127.0.0.1:12400`, so the browser you open the URL in must be running on the same machine as the CLI.
+*Note:* The authentication callback is served on `127.0.0.1`, on the first free port between `12400` and `12404`. The browser you open the URL in must run on the same machine as the CLI.
 
 #### Getting Your API Key
 

cloudsmith_cli/cli/commands/auth.py

@@ -13,7 +13,37 @@
 
 # Authentication server configuration
 AUTH_SERVER_HOST = "127.0.0.1"
-AUTH_SERVER_PORT = 12400
+AUTH_REDIRECT_HOST = "localhost"
+AUTH_SERVER_PORTS = (12400, 12401, 12402, 12403, 12404)
+
+
+def _create_auth_server(opts, owner, session, enable_token_creation, profile):
+    """Create an authentication server on the first available callback port."""
+    last_error = None
+
+    for port in AUTH_SERVER_PORTS:
+        try:
+            server = AuthenticationWebServer(
+                (AUTH_SERVER_HOST, port),
+                AuthenticationWebRequestHandler,
+                owner=owner,
+                session=session,
+                debug=opts.debug,
+                refresh_api_on_success=enable_token_creation,
+                api_opts=opts.api_config,
+                profile=profile,
+            )
+            return server, port
+        except OSError as exc:
+            last_error = exc
+
+    ports = ", ".join(str(port) for port in AUTH_SERVER_PORTS)
+    raise click.ClickException(
+        "Could not start the authentication callback server. "
+        f"Every candidate port is unavailable: {ports}. "
+        f"The last error was: {last_error}. "
+        "Stop the process that holds one of these ports, then try again."
+    ) from last_error
 
 
 def _perform_saml_authentication(
@@ -28,50 +58,52 @@ def _perform_saml_authentication(
     session = create_configured_session(opts)
     api_host = opts.api_config.host
 
-    idp_url = get_idp_url(api_host, owner, session=session)
-
-    click.echo(
-        f"Your Workspace's SAML IDP URL is: {click.style(idp_url, bold=True)}",
-        err=use_stderr,
+    auth_server, port = _create_auth_server(
+        opts, owner, session, enable_token_creation, profile
     )
-    click.echo(err=use_stderr)
+    redirect_url = f"http://{AUTH_REDIRECT_HOST}:{port}"
+
+    try:
+        idp_url = get_idp_url(
+            api_host, owner, redirect_url=redirect_url, session=session
+        )
 
-    if no_browser:
         click.echo(
-            "Skipping automatic browser launch (--no-browser). "
-            "Please open the URL above manually to continue.",
+            f"Your Workspace's SAML IDP URL is: {click.style(idp_url, bold=True)}",
             err=use_stderr,
         )
-    else:
-        try:
-            browser_opened = webbrowser.open(idp_url)
-        except Exception:
-            # Browser launch failures vary by platform (webbrowser.Error on
-            # Cygwin/headless, anything else elsewhere), so catch broadly and
-            # fall back to the manual URL rather than crashing.
-            browser_opened = False
-
-        if not browser_opened:
+        click.echo(err=use_stderr)
+
+        if no_browser:
             click.echo(
-                "Couldn't open a browser automatically. "
+                "Skipping automatic browser launch (--no-browser). "
                 "Please open the URL above manually to continue.",
                 err=use_stderr,
             )
+        else:
+            try:
+                browser_opened = webbrowser.open(idp_url)
+            except Exception:
+                # Browser launch failures vary by platform (webbrowser.Error on
+                # Cygwin/headless, anything else elsewhere), so catch broadly and
+                # fall back to the manual URL rather than crashing.
+                browser_opened = False
+
+            if not browser_opened:
+                click.echo(
+                    "Couldn't open a browser automatically. "
+                    "Please open the URL above manually to continue.",
+                    err=use_stderr,
+                )
 
-    click.echo("Starting webserver to begin authentication ... ", err=use_stderr)
-
-    auth_server = AuthenticationWebServer(
-        (AUTH_SERVER_HOST, AUTH_SERVER_PORT),
-        AuthenticationWebRequestHandler,
-        owner=owner,
-        session=session,
-        debug=opts.debug,
-        refresh_api_on_success=enable_token_creation,
-        api_opts=opts.api_config,
-        profile=profile,
-    )
+        click.echo(
+            f"Waiting for the authentication callback on port {port} ... ",
+            err=use_stderr,
+        )
 
-    auth_server.handle_request()
+        auth_server.handle_request()
+    finally:
+        auth_server.server_close()
 
 
 @main.command(aliases=["auth"])

cloudsmith_cli/cli/saml.py

@@ -34,11 +34,11 @@ def create_configured_session(opts):
     return session
 
 
-def get_idp_url(api_host, owner, session):
+def get_idp_url(api_host, owner, session, redirect_url):
     org_saml_url = "{api_host}/orgs/{owner}/saml/?{params}".format(
         api_host=api_host,
         owner=owner,
-        params=urlencode({"redirect_url": "http://localhost:12400"}),
+        params=urlencode({"redirect_url": redirect_url}),
     )
 
     org_saml_response = session.get(org_saml_url, timeout=30)

cloudsmith_cli/cli/tests/commands/test_auth.py

@@ -2,7 +2,7 @@
 
 import json
 import webbrowser
-from unittest.mock import MagicMock, patch
+from unittest.mock import ANY, MagicMock, patch
 
 import pytest
 
@@ -121,6 +121,71 @@ def test_auth_command_invokes_webserver(
         # Verify AuthenticationWebServer was called
         mock_auth_server.assert_called_once()
 
+    def test_auth_command_uses_first_available_redirect_port(
+        self,
+        runner,
+        mock_saml_session,
+        mock_get_idp_url,
+        mock_webbrowser,
+        mock_auth_server,
+    ):
+        """Verify occupied callback ports are skipped without user interaction."""
+        auth_server = MagicMock()
+        mock_auth_server.side_effect = [
+            OSError("port unavailable"),
+            OSError("port unavailable"),
+            auth_server,
+        ]
+
+        result = runner.invoke(
+            authenticate,
+            ["--owner", "testorg", "--no-browser"],
+            catch_exceptions=False,
+        )
+
+        assert result.exit_code == 0
+        assert [call.args[0] for call in mock_auth_server.call_args_list] == [
+            ("127.0.0.1", 12400),
+            ("127.0.0.1", 12401),
+            ("127.0.0.1", 12402),
+        ]
+        mock_get_idp_url.assert_called_once_with(
+            ANY,
+            "testorg",
+            redirect_url="http://localhost:12402",
+            session=mock_saml_session.return_value,
+        )
+        auth_server.handle_request.assert_called_once()
+        auth_server.server_close.assert_called_once()
+
+    def test_auth_command_fails_after_all_redirect_ports_are_unavailable(
+        self,
+        runner,
+        mock_saml_session,
+        mock_get_idp_url,
+        mock_webbrowser,
+        mock_auth_server,
+    ):
+        """Verify authentication fails only after every callback port is tried."""
+        mock_auth_server.side_effect = OSError("port unavailable")
+
+        result = runner.invoke(
+            authenticate,
+            ["--owner", "testorg", "--no-browser"],
+        )
+
+        assert result.exit_code != 0
+        assert [call.args[0] for call in mock_auth_server.call_args_list] == [
+            ("127.0.0.1", 12400),
+            ("127.0.0.1", 12401),
+            ("127.0.0.1", 12402),
+            ("127.0.0.1", 12403),
+            ("127.0.0.1", 12404),
+        ]
+        mock_get_idp_url.assert_not_called()
+        assert isinstance(result.exception, SystemExit)
+        assert "12400, 12401, 12402, 12403, 12404" in result.output
+
     def test_auth_command_opens_browser(
         self,
         runner,

cloudsmith_cli/cli/tests/test_saml.py

@@ -31,7 +31,12 @@ def test_get_idp_url(self, mock_response, mock_session):
         mock_session.get.return_value = mock_response
 
         assert (
-            get_idp_url(self.api_host, "test_org", session=mock_session)
+            get_idp_url(
+                self.api_host,
+                "test_org",
+                redirect_url="http://localhost:12400",
+                session=mock_session,
+            )
             == "response_redirect_url"
         )
         mock_session.get.assert_called_once_with(
@@ -50,15 +55,20 @@ def test_get_idp_url_with_request_error(self, mock_response, mock_session):
         )
 
         with pytest.raises(ApiException) as exc:
-            get_idp_url(self.api_host, "test_org", session=mock_session)
-
-            assert exc == ApiException(
-                status=500, headers={"foo": "bar"}, body="Error body"
-            )
-            mock_session.get.assert_called_once_with(
-                f"{self.api_host}/orgs/test_org/saml/?{self.query_params}", timeout=30
+            get_idp_url(
+                self.api_host,
+                "test_org",
+                redirect_url="http://localhost:12400",
+                session=mock_session,
             )
 
+        assert exc.value.status == 500
+        assert exc.value.headers == {"foo": "bar"}
+        assert exc.value.body == "Error body"
+        mock_session.get.assert_called_once_with(
+            f"{self.api_host}/orgs/test_org/saml/?{self.query_params}", timeout=30
+        )
+
     def test_error_carries_the_api_detail(self, mock_response, mock_session):
         import requests