Implement claims challenge error for AzureCliCredential get_token and get_token_info methods

Co-authored-by: xiangyan99 <[email protected]>

Author: Copilot

sdk/identity/azure-identity/azure/identity/_credentials/azure_cli.py

@@ -90,7 +90,7 @@ def close(self) -> None:
     def get_token(
         self,
         *scopes: str,
-        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        claims: Optional[str] = None,
         tenant_id: Optional[str] = None,
         **kwargs: Any,
     ) -> AccessToken:
@@ -102,16 +102,19 @@ def get_token(
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/entra/identity-platform/scopes-oidc.
-        :keyword str claims: not used by this credential; any value provided will be ignored.
+        :keyword str claims: additional claims required in the token. This credential does not support claims challenges.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
 
-        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI.
+        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI,
+          or when claims challenge is provided.
         :raises ~azure.core.exceptions.ClientAuthenticationError: the credential invoked the Azure CLI but didn't
           receive an access token.
         """
+        if claims:
+            raise CredentialUnavailableError(f"Fail to get token, please run az login --claims-challenge {claims}")
 
         options: TokenRequestOptions = {}
         if tenant_id:
@@ -136,10 +139,15 @@ def get_token_info(self, *scopes: str, options: Optional[TokenRequestOptions] =
         :rtype: ~azure.core.credentials.AccessTokenInfo
         :return: An AccessTokenInfo instance containing information about the token.
 
-        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI.
+        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI,
+          or when claims challenge is provided.
         :raises ~azure.core.exceptions.ClientAuthenticationError: the credential invoked the Azure CLI but didn't
           receive an access token.
         """
+        if options and options.get("claims"):
+            claims = options["claims"]
+            raise CredentialUnavailableError(f"Fail to get token, please run az login --claims-challenge {claims}")
+            
         return self._get_token_base(*scopes, options=options)
 
     def _get_token_base(

sdk/identity/azure-identity/azure/identity/aio/_credentials/azure_cli.py

@@ -82,7 +82,7 @@ def __init__(
     async def get_token(
         self,
         *scopes: str,
-        claims: Optional[str] = None,  # pylint:disable=unused-argument
+        claims: Optional[str] = None,
         tenant_id: Optional[str] = None,
         **kwargs: Any,
     ) -> AccessToken:
@@ -94,15 +94,19 @@ async def get_token(
         :param str scopes: desired scope for the access token. This credential allows only one scope per request.
             For more information about scopes, see
             https://learn.microsoft.com/entra/identity-platform/scopes-oidc.
-        :keyword str claims: not used by this credential; any value provided will be ignored.
+        :keyword str claims: additional claims required in the token. This credential does not support claims challenges.
         :keyword str tenant_id: optional tenant to include in the token request.
 
         :return: An access token with the desired scopes.
         :rtype: ~azure.core.credentials.AccessToken
-        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI.
+        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI,
+          or when claims challenge is provided.
         :raises ~azure.core.exceptions.ClientAuthenticationError: the credential invoked the Azure CLI but didn't
           receive an access token.
         """
+        if claims:
+            raise CredentialUnavailableError(f"Fail to get token, please run az login --claims-challenge {claims}")
+            
         # only ProactorEventLoop supports subprocesses on Windows (and it isn't the default loop on Python < 3.8)
         if sys.platform.startswith("win") and not isinstance(asyncio.get_event_loop(), asyncio.ProactorEventLoop):
             return _SyncAzureCliCredential().get_token(*scopes, tenant_id=tenant_id, **kwargs)
@@ -130,10 +134,15 @@ async def get_token_info(self, *scopes: str, options: Optional[TokenRequestOptio
         :rtype: ~azure.core.credentials.AccessTokenInfo
         :return: An AccessTokenInfo instance containing information about the token.
 
-        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI.
+        :raises ~azure.identity.CredentialUnavailableError: the credential was unable to invoke the Azure CLI,
+          or when claims challenge is provided.
         :raises ~azure.core.exceptions.ClientAuthenticationError: the credential invoked the Azure CLI but didn't
           receive an access token.
         """
+        if options and options.get("claims"):
+            claims = options["claims"]
+            raise CredentialUnavailableError(f"Fail to get token, please run az login --claims-challenge {claims}")
+            
         # only ProactorEventLoop supports subprocesses on Windows (and it isn't the default loop on Python < 3.8)
         if sys.platform.startswith("win") and not isinstance(asyncio.get_event_loop(), asyncio.ProactorEventLoop):
             return _SyncAzureCliCredential().get_token_info(*scopes, options=options)

sdk/identity/azure-identity/tests/test_cli_credential.py

@@ -395,3 +395,56 @@ def fake_check_output(command_line, **_):
                     kwargs = {"options": kwargs}
                 token = getattr(credential, get_token_method)("scope", **kwargs)
             assert token.token == expected_token
+
+
+@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
+def test_claims_challenge_raises_error(get_token_method):
+    """The credential should raise CredentialUnavailableError when claims challenge is provided"""
+    
+    claims = "test-claims-challenge"
+    expected_message = f"Fail to get token, please run az login --claims-challenge {claims}"
+    
+    if get_token_method == "get_token":
+        with pytest.raises(CredentialUnavailableError, match=re.escape(expected_message)):
+            AzureCliCredential().get_token("scope", claims=claims)
+    else:  # get_token_info
+        with pytest.raises(CredentialUnavailableError, match=re.escape(expected_message)):
+            AzureCliCredential().get_token_info("scope", options={"claims": claims})
+
+
+@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
+def test_empty_claims_does_not_raise_error(get_token_method):
+    """The credential should not raise error when claims parameter is empty or None"""
+    
+    # Mock the CLI to avoid actual invocation
+    with mock.patch("shutil.which", return_value="az"):
+        with mock.patch(CHECK_OUTPUT, mock.Mock(return_value='{"accessToken": "token", "expiresOn": "2021-10-07 12:00:00.000000"}')):
+            
+            if get_token_method == "get_token":
+                # Test with None (default)
+                token = AzureCliCredential().get_token("scope")
+                assert token.token == "token"
+                
+                # Test with empty string
+                token = AzureCliCredential().get_token("scope", claims="")
+                assert token.token == "token"
+                
+                # Test with None explicitly
+                token = AzureCliCredential().get_token("scope", claims=None)
+                assert token.token == "token"
+            else:  # get_token_info
+                # Test with None options
+                token = AzureCliCredential().get_token_info("scope")
+                assert token.token == "token"
+                
+                # Test with empty options
+                token = AzureCliCredential().get_token_info("scope", options={})
+                assert token.token == "token"
+                
+                # Test with None claims in options
+                token = AzureCliCredential().get_token_info("scope", options={"claims": None})
+                assert token.token == "token"
+                
+                # Test with empty string claims in options 
+                token = AzureCliCredential().get_token_info("scope", options={"claims": ""})
+                assert token.token == "token"

sdk/identity/azure-identity/tests/test_cli_credential_async.py

@@ -389,3 +389,64 @@ async def fake_exec(*args, **_):
                     kwargs = {"options": kwargs}
                 token = await getattr(credential, get_token_method)("scope", **kwargs)
             assert token.token == expected_token
+
+
+@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
+async def test_claims_challenge_raises_error(get_token_method):
+    """The credential should raise CredentialUnavailableError when claims challenge is provided"""
+    
+    claims = "test-claims-challenge"
+    expected_message = f"Fail to get token, please run az login --claims-challenge {claims}"
+    
+    if get_token_method == "get_token":
+        with pytest.raises(CredentialUnavailableError, match=re.escape(expected_message)):
+            await AzureCliCredential().get_token("scope", claims=claims)
+    else:  # get_token_info
+        with pytest.raises(CredentialUnavailableError, match=re.escape(expected_message)):
+            await AzureCliCredential().get_token_info("scope", options={"claims": claims})
+
+
+@pytest.mark.parametrize("get_token_method", GET_TOKEN_METHODS)
+async def test_empty_claims_does_not_raise_error(get_token_method):
+    """The credential should not raise error when claims parameter is empty or None"""
+    
+    successful_output = json.dumps({
+        "expiresOn": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),
+        "accessToken": "access-token",
+        "subscription": "subscription",
+        "tenant": "tenant",
+        "tokenType": "Bearer",
+    })
+    
+    # Mock the CLI to avoid actual invocation
+    with mock.patch("shutil.which", return_value="az"):
+        with mock.patch(SUBPROCESS_EXEC, mock_exec(successful_output)):
+            
+            if get_token_method == "get_token":
+                # Test with None (default)
+                token = await AzureCliCredential().get_token("scope")
+                assert token.token == "access-token"
+                
+                # Test with empty string
+                token = await AzureCliCredential().get_token("scope", claims="")
+                assert token.token == "access-token"
+                
+                # Test with None explicitly
+                token = await AzureCliCredential().get_token("scope", claims=None)
+                assert token.token == "access-token"
+            else:  # get_token_info
+                # Test with None options
+                token = await AzureCliCredential().get_token_info("scope")
+                assert token.token == "access-token"
+                
+                # Test with empty options
+                token = await AzureCliCredential().get_token_info("scope", options={})
+                assert token.token == "access-token"
+                
+                # Test with None claims in options
+                token = await AzureCliCredential().get_token_info("scope", options={"claims": None})
+                assert token.token == "access-token"
+                
+                # Test with empty string claims in options 
+                token = await AzureCliCredential().get_token_info("scope", options={"claims": ""})
+                assert token.token == "access-token"