[Identity] Encode claims for certain credentials (#42885)

Subprocess-based credentials rely on the claims being base64-encoded,
however out authentication policies call get_token/get_token_info with
claims as a JSON string. Here, we ensure that the claims are encoded
before being used.

Signed-off-by: Paul Van Eck <[email protected]>

Author: pvaneck

.vscode/cspell.json

@@ -643,7 +643,8 @@
     {
       "filename": "sdk/identity/azure-identity/tests/*.py",
       "words": [
-        "infile"
+        "infile",
+        "acrs"
       ]
     },
     {

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

@@ -17,7 +17,7 @@
 from azure.core.exceptions import ClientAuthenticationError
 
 from .. import CredentialUnavailableError
-from .._internal import resolve_tenant, within_dac, validate_tenant_id, validate_scope
+from .._internal import encode_base64, resolve_tenant, within_dac, validate_tenant_id, validate_scope
 from .._internal.decorators import log_get_token
 
 
@@ -184,7 +184,7 @@ def _get_token_base(
         if tenant:
             command_args += ["--tenant-id", tenant]
         if claims:
-            command_args += ["--claims", claims]
+            command_args += ["--claims", encode_base64(claims)]
         output = _run_command(command_args, self._process_timeout)
 
         token = parse_token(output)

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

@@ -18,6 +18,7 @@
 from .. import CredentialUnavailableError
 from .._internal import (
     _scopes_to_resource,
+    encode_base64,
     resolve_tenant,
     within_dac,
     validate_tenant_id,
@@ -155,8 +156,8 @@ def _get_token_base(
         self, *scopes: str, options: Optional[TokenRequestOptions] = None, **kwargs: Any
     ) -> AccessTokenInfo:
         # Check for claims challenge first
-        if options and options.get("claims"):
-            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=options.get("claims"))
+        if options and "claims" in options and options["claims"]:
+            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=encode_base64(options["claims"]))
 
             # Add tenant if provided in options
             if options.get("tenant_id"):

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

@@ -15,6 +15,7 @@
 from .. import CredentialUnavailableError
 from .._internal import (
     _scopes_to_resource,
+    encode_base64,
     resolve_tenant,
     within_dac,
     validate_tenant_id,
@@ -183,8 +184,8 @@ def _get_token_base(
     ) -> AccessTokenInfo:
 
         # Check if claims challenge is provided
-        if options and options.get("claims"):
-            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=options.get("claims"))
+        if options and "claims" in options and options["claims"]:
+            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=encode_base64(options["claims"]))
             if options.get("tenant_id"):
                 error_message += f" -Tenant {options.get('tenant_id')}"
             raise CredentialUnavailableError(message=error_message)

sdk/identity/azure-identity/azure/identity/_internal/__init__.py

@@ -9,6 +9,7 @@
 from .decorators import wrap_exceptions
 from .interactive import InteractiveCredential
 from .utils import (
+    encode_base64,
     get_default_authority,
     normalize_authority,
     process_credential_exclusions,
@@ -46,6 +47,7 @@ def _scopes_to_resource(*scopes) -> str:
     "AadClientBase",
     "AuthCodeRedirectServer",
     "AadClientCertificate",
+    "encode_base64",
     "get_default_authority",
     "InteractiveCredential",
     "normalize_authority",

sdk/identity/azure-identity/azure/identity/_internal/utils.py

@@ -2,6 +2,7 @@
 # Copyright (c) Microsoft Corporation.
 # Licensed under the MIT License.
 # ------------------------------------
+import base64
 import os
 import platform
 import logging
@@ -223,3 +224,8 @@ def is_wsl() -> bool:
     platform_name = getattr(uname, "system", uname[0]).lower()
     release = getattr(uname, "release", uname[2]).lower()
     return platform_name == "linux" and "microsoft" in release
+
+
+def encode_base64(s: str) -> str:
+    encoded = base64.b64encode(s.encode("utf-8"))
+    return encoded.decode("utf-8")

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

@@ -26,7 +26,7 @@
     sanitize_output,
     extract_cli_error_message,
 )
-from ..._internal import resolve_tenant, within_dac, validate_tenant_id, validate_scope
+from ..._internal import encode_base64, resolve_tenant, within_dac, validate_tenant_id, validate_scope
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -176,7 +176,7 @@ async def _get_token_base(
         if tenant:
             command_args += ["--tenant-id", tenant]
         if claims:
-            command_args += ["--claims", claims]
+            command_args += ["--claims", encode_base64(claims)]
         output = await _run_command(command_args, self._process_timeout)
 
         token = parse_token(output)

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

@@ -27,6 +27,7 @@
 )
 from ..._internal import (
     _scopes_to_resource,
+    encode_base64,
     resolve_tenant,
     within_dac,
     validate_tenant_id,
@@ -149,8 +150,8 @@ async def _get_token_base(
         self, *scopes: str, options: Optional[TokenRequestOptions] = None, **kwargs: Any
     ) -> AccessTokenInfo:
         # Check for claims challenge first
-        if options and options.get("claims"):
-            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=options.get("claims"))
+        if options and "claims" in options and options["claims"]:
+            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=encode_base64(options["claims"]))
 
             # Add tenant if provided in options
             if options.get("tenant_id"):

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

@@ -18,7 +18,7 @@
     parse_token,
     CLAIMS_UNSUPPORTED_ERROR,
 )
-from ..._internal import resolve_tenant, validate_tenant_id, validate_scope
+from ..._internal import encode_base64, resolve_tenant, validate_tenant_id, validate_scope
 
 
 class AzurePowerShellCredential(AsyncContextManager):
@@ -124,8 +124,8 @@ async def _get_token_base(
     ) -> AccessTokenInfo:
 
         # Check if claims challenge is provided
-        if options and options.get("claims"):
-            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=options.get("claims"))
+        if options and "claims" in options and options["claims"]:
+            error_message = CLAIMS_UNSUPPORTED_ERROR.format(claims_value=encode_base64(options["claims"]))
             if options.get("tenant_id"):
                 error_message += f" -Tenant {options.get('tenant_id')}"
             raise CredentialUnavailableError(message=error_message)

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

@@ -343,7 +343,7 @@ def fake_check_output(command_line, **_):
 def test_claims_challenge_raises_error(get_token_method):
     """The credential should raise CredentialUnavailableError when claims challenge is provided"""
 
-    claims = "test-claims-challenge"
+    claims = '{"access_token":{"acrs":{"essential":true,"values":["p1"]}}}'
     credential = AzureDeveloperCliCredential()
 
     expected_message = "Suggestion: re-authentication required, run `azd auth login` to acquire a new token."
@@ -405,15 +405,16 @@ def test_empty_claims_does_not_raise_error(get_token_method):
 def test_claims_command_line_argument(get_token_method):
     """The credential should pass claims as --claims argument to azd command"""
 
-    claims = "test-claims-challenge"
+    claims = '{"access_token":{"acrs":{"essential":true,"values":["p1"]}}}'
+    expected_encoded_claims = "eyJhY2Nlc3NfdG9rZW4iOnsiYWNycyI6eyJlc3NlbnRpYWwiOnRydWUsInZhbHVlcyI6WyJwMSJdfX19"
     access_token = "access token"
     expected_expires_on = 1602015811
 
     def fake_check_output(command_line, **kwargs):
         # Verify that claims are passed as --claims argument
         assert "--claims" in command_line
         claims_index = command_line.index("--claims")
-        assert command_line[claims_index + 1] == claims
+        assert command_line[claims_index + 1] == expected_encoded_claims
 
         return json.dumps(
             {