You may close this tab and return to your terminal.
') + if ('code' in params and 'state' in params) or 'error' in params: + callbacks.put(params) + threading.Thread(target=self.server.shutdown, daemon=True).start() + + def log_message(self, fmt, *args): + pass + + server = http.server.HTTPServer(('localhost', 0), _CallbackHandler) + port = server.server_address[1] + threading.Thread(target=server.serve_forever, daemon=True).start() + + # response_mode='query' is intentional — we need the code in the URL for the paste fallback + with warnings.catch_warnings(): + warnings.simplefilter('ignore') + flow = self._msal_app.initiate_auth_code_flow( + scopes, redirect_uri='http://localhost:{}'.format(port), + prompt='select_account', claims_challenge=claims_challenge, + response_mode='query') + + input('Press Enter to open the browser for sign-in...') + webbrowser.open(flow['auth_uri']) + logger.warning("If the browser did not open, visit: %s", flow['auth_uri']) + + url = input('After signing in, press Enter to continue ' + '(or paste the redirected URL if your browser shows a connection error): ').strip() + + try: + if url: + auth_response = _parse_codespaces_auth_response(url) + else: + try: + auth_response = callbacks.get(timeout=300) + except queue.Empty: + raise CLIError('Login timed out waiting for the redirected response. ' + 'Please try again, or paste the redirected URL.') + finally: + server.shutdown() + server.server_close() + result = self._msal_app.acquire_token_by_auth_code_flow(flow, auth_response, scopes=scopes) + return check_result(result) + def login_with_device_code(self, scopes, claims_challenge=None): flow = self._msal_app.initiate_device_flow(scopes, claims_challenge=claims_challenge) if "user_code" not in flow: @@ -438,6 +500,26 @@ def _try_remove(path): pass +def _parse_codespaces_auth_response(redirected_url): + if not redirected_url: + raise CLIError('No redirected URL was provided.') + + parsed = urlparse(redirected_url) + params = parse_qs(parsed.query or '', keep_blank_values=True) + + if 'error' in params: + description = params.get('error_description', [''])[0] + raise CLIError('Authentication failed: {} {}'.format(params.get('error', [''])[0], description).strip()) + + code = params.get('code', [''])[0] + state = params.get('state', [''])[0] + + if not code or not state: + raise CLIError('Redirected URL does not include required parameters "code" and "state".') + + return {k: v[0] for k, v in params.items() if v and v[0] != ''} + + def get_environment_credential(): # A temporary workaround used by rdbms module to use environment credential. # TODO: Integrate with Identity and utilize MSAL HTTP and token cache to officially implement diff --git a/src/azure-cli-core/azure/cli/core/tests/test_profile.py b/src/azure-cli-core/azure/cli/core/tests/test_profile.py index d62464ca5d3..dd0f620888c 100644 --- a/src/azure-cli-core/azure/cli/core/tests/test_profile.py +++ b/src/azure-cli-core/azure/cli/core/tests/test_profile.py @@ -15,6 +15,7 @@ _transform_subscription_for_multiapi, _TENANT_LEVEL_ACCOUNT_NAME) from azure.cli.core.azclierror import AuthenticationError +from azure.cli.core.auth.identity import _parse_codespaces_auth_response from azure.cli.core.auth.util import AccessToken from azure.cli.core.mock import DummyCli from azure.mgmt.resource.subscriptions.models import \ @@ -379,15 +380,13 @@ def test_login_fallback_to_device_code_no_browser(self, can_launch_browser_mock, @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) + @mock.patch('azure.cli.core.auth.identity.Identity.login_with_auth_code_for_codespaces', 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_fallback_to_device_code_github_codespaces(self, can_launch_browser_mock, is_github_codespaces_mock, - login_with_device_code_mock, get_user_credential_mock, - create_subscription_client_mock): - # GitHub Codespaces does support launching a browser (actually a new tab), - # so we mock can_launch_browser to True. - login_with_device_code_mock.return_value = self.user_identity_mock + def test_login_with_auth_code_github_codespaces(self, can_launch_browser_mock, is_github_codespaces_mock, + login_with_auth_code_for_codespaces_mock, + get_user_credential_mock, create_subscription_client_mock): + login_with_auth_code_for_codespaces_mock.return_value = self.user_identity_mock cli = DummyCli() mock_subscription_client = mock.MagicMock() @@ -397,12 +396,61 @@ 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() + login_with_auth_code_for_codespaces_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_device_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_does_not_use_device_code_in_codespaces(self, can_launch_browser_mock, + is_github_codespaces_mock, + login_with_device_code_mock, + get_user_credential_mock, + create_subscription_client_mock): + """Codespaces detection must route to auth_code flow, not device code.""" + 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 + + storage_mock = {'subscriptions': None} + profile = Profile(cli_ctx=cli, storage=storage_mock) + with mock.patch('azure.cli.core.auth.identity.Identity.login_with_auth_code_for_codespaces', + autospec=True, return_value=self.user_identity_mock): + profile.login(True, None, None, False, None, use_device_code=False, allow_no_subscriptions=False) + + login_with_device_code_mock.assert_not_called() + + def test_parse_codespaces_auth_response_valid(self): + url = 'http://localhost:12345?code=abc123&state=xyz&session_state=ss' + result = _parse_codespaces_auth_response(url) + self.assertEqual('abc123', result['code']) + self.assertEqual('xyz', result['state']) + + def test_parse_codespaces_auth_response_error(self): + url = 'http://localhost:12345?error=access_denied&error_description=User+cancelled' + from knack.util import CLIError + with self.assertRaises(CLIError) as ctx: + _parse_codespaces_auth_response(url) + self.assertIn('access_denied', str(ctx.exception)) + + def test_parse_codespaces_auth_response_missing_code(self): + url = 'http://localhost:12345?state=xyz' + from knack.util import CLIError + with self.assertRaises(CLIError): + _parse_codespaces_auth_response(url) + + def test_parse_codespaces_auth_response_empty(self): + from knack.util import CLIError + with self.assertRaises(CLIError): + _parse_codespaces_auth_response('') + @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)