[Profile] az login: Add --redirect-port for browser authentication

src/azure-cli-core/azure/cli/core/_profile.py

@@ -154,29 +154,33 @@ def login(self,
               show_progress=False,
               claims_challenge=None,
               skip_subscription_discovery=False,
-              subscription=None):
+              subscription=None,
+              redirect_port=None):
         """
         For service principal, `password` is a dict returned by ServicePrincipalAuth.build_credential
         """
         if not scopes:
             scopes = self._arm_scope
 
-        identity = _create_identity_instance(self.cli_ctx, self._authority, tenant_id=tenant)
+        identity = _create_identity_instance(
+            self.cli_ctx, self._authority, tenant_id=tenant,
+            enable_broker_on_windows=False if redirect_port is not None else None)
 
         user_identity = None
         if interactive:
-            if not use_device_code and not can_launch_browser():
+            if not use_device_code and redirect_port is None and not can_launch_browser():
                 logger.info('No web browser is available. Fall back to device code.')
                 use_device_code = True
 
-            if not use_device_code and is_github_codespaces():
+            if not use_device_code and redirect_port is None and is_github_codespaces():
                 logger.info('GitHub Codespaces is detected. Fall back to device code.')
                 use_device_code = True
 
             if use_device_code:
                 user_identity = identity.login_with_device_code(scopes=scopes, claims_challenge=claims_challenge)
             else:
-                user_identity = identity.login_with_auth_code(scopes=scopes, claims_challenge=claims_challenge)
+                user_identity = identity.login_with_auth_code(scopes=scopes, claims_challenge=claims_challenge,
+                                                              redirect_port=redirect_port)
         else:
             if not is_service_principal:
                 user_identity = identity.login_with_username_password(username, password, scopes=scopes)
@@ -964,7 +968,7 @@ def _transform_subscription_for_multiapi(s, s_dict):
             s_dict[_MANAGED_BY_TENANTS] = [{_TENANT_ID: t.tenant_id} for t in s.managed_by_tenants]
 
 
-def _create_identity_instance(cli_ctx, authority, tenant_id=None, client_id=None):
+def _create_identity_instance(cli_ctx, authority, tenant_id=None, client_id=None, enable_broker_on_windows=None):
     """Lazily import and create Identity instance to avoid unnecessary imports."""
     from .auth.identity import Identity
     from .util import should_encrypt_token_cache
@@ -974,7 +978,8 @@ def _create_identity_instance(cli_ctx, authority, tenant_id=None, client_id=None
     use_msal_http_cache = cli_ctx.config.getboolean('core', 'use_msal_http_cache', fallback=True)
 
     # On Windows, use core.enable_broker_on_windows=false to disable broker (WAM) for authentication.
-    enable_broker_on_windows = cli_ctx.config.getboolean('core', 'enable_broker_on_windows', fallback=True)
+    if enable_broker_on_windows is None:
+        enable_broker_on_windows = cli_ctx.config.getboolean('core', 'enable_broker_on_windows', fallback=True)
     from .telemetry import set_broker_info
     set_broker_info(enable_broker_on_windows)
 

src/azure-cli-core/azure/cli/core/auth/identity.py

@@ -3,12 +3,14 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import errno
 import json
 import os
 import re
 import sys
 
 from azure.cli.core._environment import get_config_dir
+from azure.cli.core.azclierror import ClientRequestError, InvalidArgumentValueError
 from knack.log import get_logger
 from knack.util import CLIError
 from msal import PublicClientApplication, ConfidentialClientApplication
@@ -145,7 +147,7 @@ def _service_principal_store(self):
             Identity._service_principal_store_instance = ServicePrincipalStore(store)
         return Identity._service_principal_store_instance
 
-    def login_with_auth_code(self, scopes, claims_challenge=None):
+    def login_with_auth_code(self, scopes, claims_challenge=None, redirect_port=None):
         # Emit a warning to inform that a browser is opened.
         # Only show the path part of the URL and hide the query string.
 
@@ -161,14 +163,27 @@ def _prompt_launching_ui(ui=None, **_):
         from .util import read_response_templates
         success_template, error_template = read_response_templates()
 
-        # For AAD, use port 0 to let the system choose arbitrary unused ephemeral port to avoid port collision
-        # on port 8400 from the old design. However, ADFS only allows port 8400.
-        result = self._msal_app.acquire_token_interactive(
-            scopes, prompt='select_account', port=8400 if self._is_adfs else None,
-            success_template=success_template, error_template=error_template,
-            parent_window_handle=self._msal_app.CONSOLE_WINDOW_HANDLE, on_before_launching_ui=_prompt_launching_ui,
-            enable_msa_passthrough=True,
-            claims_challenge=claims_challenge)
+        if self._is_adfs and redirect_port not in (None, 8400):
+            raise InvalidArgumentValueError('--redirect-port must be 8400 when authenticating with ADFS.')
+
+        # For AAD, use port 0 to let the system choose an unused ephemeral port unless one is explicitly requested.
+        # ADFS only allows port 8400.
+        port = redirect_port if redirect_port is not None else (8400 if self._is_adfs else None)
+        try:
+            result = self._msal_app.acquire_token_interactive(
+                scopes, prompt='select_account', port=port,
+                success_template=success_template, error_template=error_template,
+                parent_window_handle=self._msal_app.CONSOLE_WINDOW_HANDLE, on_before_launching_ui=_prompt_launching_ui,
+                enable_msa_passthrough=True,
+                claims_challenge=claims_challenge)
+        except (OSError, ValueError) as ex:
+            is_bind_error = (isinstance(ex, OSError) and ex.errno in (errno.EADDRINUSE, errno.EACCES))
+            is_wrapped_permission_error = isinstance(ex, ValueError) and isinstance(ex.__context__, PermissionError)
+            if redirect_port is not None and (is_bind_error or is_wrapped_permission_error):
+                raise ClientRequestError(
+                    "Redirect port {} is unavailable.".format(redirect_port),
+                    recommendation="Free the port or choose another value for --redirect-port.") from ex
+            raise
         return check_result(result)
 
     def login_with_device_code(self, scopes, claims_challenge=None):

src/azure-cli-core/azure/cli/core/auth/tests/test_identity.py

@@ -3,13 +3,15 @@
 # Licensed under the MIT License. See License.txt in the project root for license information.
 # --------------------------------------------------------------------------------------------
 
+import errno
 import os
 import re
 import unittest
 from unittest import mock
 
 from azure.cli.core.auth.identity import (Identity, ServicePrincipalAuth, ServicePrincipalStore,
                                           _get_authority_url)
+from azure.cli.core.azclierror import ClientRequestError, InvalidArgumentValueError
 from knack.util import CLIError
 
 # CERTIFICATE section in sp_cert.pem
@@ -45,6 +47,81 @@
 
 class TestIdentity(unittest.TestCase):
 
+    @mock.patch("azure.cli.core.auth.util.read_response_templates", return_value=("success", "error"))
+    def test_login_with_auth_code_redirect_port(self, _):
+        identity = Identity('https://login.microsoftonline.com')
+        identity._msal_app_instance = mock.MagicMock()
+        identity._msal_app_instance.acquire_token_interactive.return_value = {'access_token': 'test_token'}
+
+        identity.login_with_auth_code(['scope'], redirect_port=4242)
+
+        self.assertEqual(4242, identity._msal_app_instance.acquire_token_interactive.call_args.kwargs['port'])
+
+    @mock.patch("azure.cli.core.auth.util.read_response_templates", return_value=("success", "error"))
+    def test_login_with_auth_code_default_ports(self, _):
+        for is_adfs, expected_port in ((False, None), (True, 8400)):
+            with self.subTest(is_adfs=is_adfs):
+                identity = Identity('https://login.microsoftonline.com')
+                identity._is_adfs = is_adfs
+                identity._msal_app_instance = mock.MagicMock()
+                identity._msal_app_instance.acquire_token_interactive.return_value = {'access_token': 'test_token'}
+
+                identity.login_with_auth_code(['scope'])
+
+                self.assertEqual(
+                    expected_port, identity._msal_app_instance.acquire_token_interactive.call_args.kwargs['port'])
+
+    @mock.patch("azure.cli.core.auth.util.read_response_templates", return_value=("success", "error"))
+    def test_login_with_auth_code_redirect_port_adfs(self, _):
+        identity = Identity('https://login.microsoftonline.com')
+        identity._is_adfs = True
+        identity._msal_app_instance = mock.MagicMock()
+
+        with self.assertRaisesRegex(InvalidArgumentValueError, '--redirect-port must be 8400'):
+            identity.login_with_auth_code(['scope'], redirect_port=4242)
+
+        identity._msal_app_instance.acquire_token_interactive.assert_not_called()
+        identity._msal_app_instance.acquire_token_interactive.return_value = {'access_token': 'test_token'}
+
+        identity.login_with_auth_code(['scope'], redirect_port=8400)
+
+        self.assertEqual(8400, identity._msal_app_instance.acquire_token_interactive.call_args.kwargs['port'])
+
+    @mock.patch("azure.cli.core.auth.util.read_response_templates", return_value=("success", "error"))
+    def test_login_with_auth_code_redirect_port_unavailable(self, _):
+        identity = Identity('https://login.microsoftonline.com')
+        identity._msal_app_instance = mock.MagicMock()
+        bind_error = OSError(errno.EADDRINUSE, 'Address already in use')
+        identity._msal_app_instance.acquire_token_interactive.side_effect = bind_error
+
+        with self.assertRaisesRegex(ClientRequestError, 'Redirect port 4242 is unavailable'):
+            identity.login_with_auth_code(['scope'], redirect_port=4242)
+
+    @mock.patch("azure.cli.core.auth.util.read_response_templates", return_value=("success", "error"))
+    def test_login_with_auth_code_redirect_port_permission_denied(self, _):
+        identity = Identity('https://login.microsoftonline.com')
+        identity._msal_app_instance = mock.MagicMock()
+        try:
+            try:
+                raise PermissionError(errno.EACCES, 'Permission denied')
+            except PermissionError:
+                raise ValueError("Can't listen on port 4242. You may try port 0.")
+        except ValueError as ex:
+            bind_error = ex
+        identity._msal_app_instance.acquire_token_interactive.side_effect = bind_error
+
+        with self.assertRaisesRegex(ClientRequestError, 'Redirect port 4242 is unavailable'):
+            identity.login_with_auth_code(['scope'], redirect_port=4242)
+
+    @mock.patch("azure.cli.core.auth.util.read_response_templates", return_value=("success", "error"))
+    def test_login_with_auth_code_unrelated_value_error(self, _):
+        identity = Identity('https://login.microsoftonline.com')
+        identity._msal_app_instance = mock.MagicMock()
+        identity._msal_app_instance.acquire_token_interactive.side_effect = ValueError('unrelated')
+
+        with self.assertRaisesRegex(ValueError, 'unrelated'):
+            identity.login_with_auth_code(['scope'], redirect_port=4242)
+
     @mock.patch("azure.cli.core.auth.identity.ServicePrincipalStore.save_entry")
     @mock.patch("msal.application.ConfidentialClientApplication.acquire_token_for_client")
     @mock.patch("msal.application.ConfidentialClientApplication.__init__", return_value=None)

src/azure-cli-core/azure/cli/core/tests/test_profile.py

@@ -329,10 +329,39 @@ def test_login_with_auth_code(self, can_launch_browser_mock, login_with_auth_cod
 
         storage_mock = {'subscriptions': None}
         profile = Profile(cli_ctx=cli, storage=storage_mock)
-        subs = profile.login(True, None, None, False, None, use_device_code=False, allow_no_subscriptions=False)
+        with mock.patch.object(cli.config, 'getboolean', return_value=True):
+            subs = profile.login(True, None, None, False, None, use_device_code=False, allow_no_subscriptions=False)
 
         # assert
         login_with_auth_code_mock.assert_called_once()
+        self.assertIsNone(login_with_auth_code_mock.call_args.kwargs['redirect_port'])
+        self.assertTrue(login_with_auth_code_mock.call_args.args[0]._enable_broker_on_windows)
+        get_user_credential_mock.assert_called()
+        self.assertEqual(self.subscription1_with_tenant_info_output, subs)
+
+    @mock.patch('azure.cli.core._profile.SubscriptionFinder._create_subscription_client', autospec=True)
+    @mock.patch('azure.cli.core.auth.identity.Identity.get_user_credential', autospec=True)
+    @mock.patch('azure.cli.core.auth.identity.Identity.login_with_auth_code', autospec=True)
+    @mock.patch('azure.cli.core._profile.can_launch_browser', autospec=True, return_value=False)
+    def test_login_with_auth_code_redirect_port_without_browser(self, can_launch_browser_mock,
+                                                                login_with_auth_code_mock,
+                                                                get_user_credential_mock,
+                                                                create_subscription_client_mock):
+        login_with_auth_code_mock.return_value = self.user_identity_mock
+
+        cli = DummyCli()
+        mock_subscription_client = mock.MagicMock()
+        mock_subscription_client.tenants.list.return_value = [TenantStub(self.tenant_id)]
+        mock_subscription_client.subscriptions.list.return_value = [deepcopy(self.subscription1_raw)]
+        create_subscription_client_mock.return_value = mock_subscription_client
+
+        profile = Profile(cli_ctx=cli, storage={'subscriptions': None})
+        with mock.patch.object(cli.config, 'getboolean', return_value=True):
+            subs = profile.login(True, None, None, False, None, redirect_port=4242)
+
+        login_with_auth_code_mock.assert_called_once()
+        self.assertEqual(4242, login_with_auth_code_mock.call_args.kwargs['redirect_port'])
+        self.assertFalse(login_with_auth_code_mock.call_args.args[0]._enable_broker_on_windows)
         get_user_credential_mock.assert_called()
         self.assertEqual(self.subscription1_with_tenant_info_output, subs)
 
@@ -373,7 +402,7 @@ def test_login_fallback_to_device_code_no_browser(self, can_launch_browser_mock,
 
         storage_mock = {'subscriptions': None}
         profile = Profile(cli_ctx=cli, storage=storage_mock)
-        subs = profile.login(True, None, None, False, None, use_device_code=True, allow_no_subscriptions=False)
+        subs = profile.login(True, None, None, False, None, use_device_code=False, allow_no_subscriptions=False)
 
         # assert
         login_with_device_code_mock.assert_called_once()
@@ -399,12 +428,37 @@ def test_login_fallback_to_device_code_github_codespaces(self, can_launch_browse
 
         storage_mock = {'subscriptions': None}
         profile = Profile(cli_ctx=cli, storage=storage_mock)
-        subs = profile.login(True, None, None, False, None, use_device_code=True, allow_no_subscriptions=False)
+        subs = profile.login(True, None, None, False, None, use_device_code=False, allow_no_subscriptions=False)
 
         # assert
         login_with_device_code_mock.assert_called_once()
         self.assertEqual(self.subscription1_with_tenant_info_output, subs)
 
+    @mock.patch('azure.cli.core._profile.SubscriptionFinder._create_subscription_client', autospec=True)
+    @mock.patch('azure.cli.core.auth.identity.Identity.get_user_credential', autospec=True)
+    @mock.patch('azure.cli.core.auth.identity.Identity.login_with_auth_code', autospec=True)
+    @mock.patch('azure.cli.core._profile.is_github_codespaces', autospec=True, return_value=True)
+    @mock.patch('azure.cli.core._profile.can_launch_browser', autospec=True, return_value=True)
+    def test_login_with_auth_code_redirect_port_github_codespaces(self, can_launch_browser_mock,
+                                                                 is_github_codespaces_mock,
+                                                                 login_with_auth_code_mock,
+                                                                 get_user_credential_mock,
+                                                                 create_subscription_client_mock):
+        login_with_auth_code_mock.return_value = self.user_identity_mock
+
+        cli = DummyCli()
+        mock_subscription_client = mock.MagicMock()
+        mock_subscription_client.tenants.list.return_value = [TenantStub(self.tenant_id)]
+        mock_subscription_client.subscriptions.list.return_value = [deepcopy(self.subscription1_raw)]
+        create_subscription_client_mock.return_value = mock_subscription_client
+
+        profile = Profile(cli_ctx=cli, storage={'subscriptions': None})
+        subs = profile.login(True, None, None, False, None, redirect_port=4242)
+
+        login_with_auth_code_mock.assert_called_once()
+        self.assertEqual(4242, login_with_auth_code_mock.call_args.kwargs['redirect_port'])
+        self.assertEqual(self.subscription1_with_tenant_info_output, subs)
+
     @mock.patch('azure.cli.core._profile.SubscriptionFinder._create_subscription_client', autospec=True)
     @mock.patch('azure.cli.core.auth.identity.Identity.get_user_credential', autospec=True)
     @mock.patch('azure.cli.core.auth.identity.Identity.login_with_device_code', autospec=True)

src/azure-cli/azure/cli/command_modules/profile/__init__.py

@@ -77,6 +77,15 @@ def load_arguments(self, command):
                        help="Use device code flow. Azure CLI will also use this if it can't launch a browser, "
                             "e.g. in remote SSH or Cloud Shell.")
 
+            # Interactive browser flow
+            c.argument('redirect_port', type=int,
+                       help='Port for the local server used by interactive browser authentication. By default, '
+                            'an available ephemeral port is selected. Valid values are from 1 to 65535. The command '
+                            'fails if the specified port is unavailable. ADFS only supports port 8400. This option is '
+                            'only supported with '
+                            'interactive browser authentication and disables WAM and automatic fallback to device '
+                            'code.')
+
             # Service principal
             c.argument('service_principal', action='store_true',
                        help='Log in with a service principal.')

src/azure-cli/azure/cli/command_modules/profile/_help.py

@@ -14,6 +14,9 @@
     By default, this command logs in with a user account.
     Azure CLI uses Web Account Manager (WAM) on Windows, and browser-based login on Linux and macOS by default.
     If WAM or a web browser is not available, Azure CLI will fall back to device code login.
+    Specifying --redirect-port forces browser-based login and disables this fallback and WAM. To sign in through SSH,
+    forward the same port from the machine running the browser and set BROWSER=echo on the remote machine to print the
+    login URL.
 
 
     [WARNING] Authentication with username and password in the command line is strongly discouraged.
@@ -35,6 +38,8 @@
 examples:
     - name: Log in interactively.
       text: az login
+    - name: Log in through SSH with a fixed callback port after forwarding that port from the machine running the browser.
+      text: BROWSER=echo az login --redirect-port 8400
     - name: Log in with username and password. This doesn't work with Microsoft accounts or accounts that have two-factor authentication enabled. Use -p=secret if the first character of the password is '-'.
       text: az login --username johndoe@contoso.com --password VerySecret
     - name: Log in with a service principal using client secret. Use --password=secret if the first character of the password is '-'.