[Identity] Use full path for CLI executables (#38606)

Also, adjustments were made to pass in subprocess arguments as a list.

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

Author: pvaneck

sdk/identity/azure-identity/CHANGELOG.md

@@ -14,6 +14,8 @@
 
 ### Other Changes
 
+- `AzureCliCredential` and `AzureDeveloperCliCredential` will now call their corresponding executables directly instead of going through the shell. ([#38606](https://github.com/Azure/azure-sdk-for-python/pull/38606))
+
 ## 1.19.0 (2024-10-08)
 
 ### Bugs Fixed

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

@@ -5,6 +5,7 @@
 
 from datetime import datetime
 import json
+import logging
 import os
 import re
 import shutil
@@ -19,12 +20,15 @@
 from .._internal import resolve_tenant, within_dac, validate_tenant_id, validate_scope
 from .._internal.decorators import log_get_token
 
+
+_LOGGER = logging.getLogger(__name__)
+
 CLI_NOT_FOUND = (
     "Azure Developer CLI could not be found. "
     "Please visit https://aka.ms/azure-dev for installation instructions and then,"
     "once installed, authenticate to your Azure account using 'azd auth login'."
 )
-COMMAND_LINE = "azd auth token --output json --scope {}"
+COMMAND_LINE = ["auth", "token", "--output", "json"]
 EXECUTABLE_NAME = "azd"
 NOT_LOGGED_IN = "Please run 'azd auth login' from a command prompt to authenticate before using this credential."
 
@@ -160,17 +164,18 @@ def _get_token_base(
         for scope in scopes:
             validate_scope(scope)
 
-        commandString = " --scope ".join(scopes)
-        command = COMMAND_LINE.format(commandString)
+        command_args = COMMAND_LINE.copy()
+        for scope in scopes:
+            command_args += ["--scope", scope]
         tenant = resolve_tenant(
             default_tenant=self.tenant_id,
             tenant_id=tenant_id,
             additionally_allowed_tenants=self._additionally_allowed_tenants,
             **kwargs,
         )
         if tenant:
-            command += " --tenant-id " + tenant
-        output = _run_command(command, self._process_timeout)
+            command_args += ["--tenant-id", tenant]
+        output = _run_command(command_args, self._process_timeout)
 
         token = parse_token(output)
         if not token:
@@ -236,15 +241,13 @@ def sanitize_output(output: str) -> str:
     return re.sub(r"\"token\": \"(.*?)(\"|$)", "****", output)
 
 
-def _run_command(command: str, timeout: int) -> str:
+def _run_command(command_args: List[str], timeout: int) -> str:
     # Ensure executable exists in PATH first. This avoids a subprocess call that would fail anyway.
-    if shutil.which(EXECUTABLE_NAME) is None:
+    azd_path = shutil.which(EXECUTABLE_NAME)
+    if not azd_path:
         raise CredentialUnavailableError(message=CLI_NOT_FOUND)
 
-    if sys.platform.startswith("win"):
-        args = ["cmd", "/c", command]
-    else:
-        args = ["/bin/sh", "-c", command]
+    args = [azd_path] + command_args
     try:
         working_directory = get_safe_working_dir()
 
@@ -257,6 +260,7 @@ def _run_command(command: str, timeout: int) -> str:
             "timeout": timeout,
         }
 
+        _LOGGER.debug("Executing subprocess with the following arguments %s", args)
         return subprocess.check_output(args, **kwargs)
     except subprocess.CalledProcessError as ex:
         # non-zero return from shell

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

@@ -6,6 +6,7 @@
 import json
 import os
 import re
+import logging
 import shutil
 import subprocess
 import sys
@@ -26,8 +27,11 @@
 from .._internal.decorators import log_get_token
 
 
+_LOGGER = logging.getLogger(__name__)
+
 CLI_NOT_FOUND = "Azure CLI not found on path"
-COMMAND_LINE = "az account get-access-token --output json --resource {}"
+# COMMAND_LINE = "account get-access-token --output json --resource {}"
+COMMAND_LINE = ["account", "get-access-token", "--output", "json"]
 EXECUTABLE_NAME = "az"
 NOT_LOGGED_IN = "Please run 'az login' to set up an account"
 
@@ -149,19 +153,19 @@ def _get_token_base(
             validate_scope(scope)
 
         resource = _scopes_to_resource(*scopes)
-        command = COMMAND_LINE.format(resource)
+        command_args = COMMAND_LINE + ["--resource", resource]
         tenant = resolve_tenant(
             default_tenant=self.tenant_id,
             tenant_id=tenant_id,
             additionally_allowed_tenants=self._additionally_allowed_tenants,
             **kwargs,
         )
         if tenant:
-            command += " --tenant " + tenant
+            command_args += ["--tenant", tenant]
 
         if self.subscription:
-            command += f' --subscription "{self.subscription}"'
-        output = _run_command(command, self._process_timeout)
+            command_args += ["--subscription", self.subscription]
+        output = _run_command(command_args, self._process_timeout)
 
         token = parse_token(output)
         if not token:
@@ -228,15 +232,13 @@ def sanitize_output(output: str) -> str:
     return re.sub(r"\"accessToken\": \"(.*?)(\"|$)", "****", output)
 
 
-def _run_command(command: str, timeout: int) -> str:
+def _run_command(command_args: List[str], timeout: int) -> str:
     # Ensure executable exists in PATH first. This avoids a subprocess call that would fail anyway.
-    if shutil.which(EXECUTABLE_NAME) is None:
+    az_path = shutil.which(EXECUTABLE_NAME)
+    if not az_path:
         raise CredentialUnavailableError(message=CLI_NOT_FOUND)
 
-    if sys.platform.startswith("win"):
-        args = ["cmd", "/c", command]
-    else:
-        args = ["/bin/sh", "-c", command]
+    args = [az_path] + command_args
     try:
         working_directory = get_safe_working_dir()
 
@@ -248,6 +250,7 @@ def _run_command(command: str, timeout: int) -> str:
             "timeout": timeout,
             "env": dict(os.environ, AZURE_CORE_NO_COLOR="true"),
         }
+        _LOGGER.debug("Executing subprocess with the following arguments %s", args)
         return subprocess.check_output(args, **kwargs)
     except subprocess.CalledProcessError as ex:
         # non-zero return from shell

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

@@ -3,6 +3,7 @@
 # Licensed under the MIT License.
 # ------------------------------------
 import asyncio
+import logging
 import os
 import shutil
 import sys
@@ -26,6 +27,9 @@
 from ..._internal import resolve_tenant, within_dac, validate_tenant_id, validate_scope
 
 
+_LOGGER = logging.getLogger(__name__)
+
+
 class AzureDeveloperCliCredential(AsyncContextManager):
     """Authenticates by requesting a token from the Azure Developer CLI.
 
@@ -153,8 +157,9 @@ async def _get_token_base(
         for scope in scopes:
             validate_scope(scope)
 
-        commandString = " --scope ".join(scopes)
-        command = COMMAND_LINE.format(commandString)
+        command_args = COMMAND_LINE.copy()
+        for scope in scopes:
+            command_args += ["--scope", scope]
         tenant = resolve_tenant(
             default_tenant=self.tenant_id,
             tenant_id=tenant_id,
@@ -163,8 +168,8 @@ async def _get_token_base(
         )
 
         if tenant:
-            command += " --tenant-id " + tenant
-        output = await _run_command(command, self._process_timeout)
+            command_args += ["--tenant-id", tenant]
+        output = await _run_command(command_args, self._process_timeout)
 
         token = parse_token(output)
         if not token:
@@ -184,19 +189,17 @@ async def close(self) -> None:
         """Calling this method is unnecessary"""
 
 
-async def _run_command(command: str, timeout: int) -> str:
+async def _run_command(command_args: List[str], timeout: int) -> str:
     # Ensure executable exists in PATH first. This avoids a subprocess call that would fail anyway.
-    if shutil.which(EXECUTABLE_NAME) is None:
+    azd_path = shutil.which(EXECUTABLE_NAME)
+    if not azd_path:
         raise CredentialUnavailableError(message=CLI_NOT_FOUND)
 
-    if sys.platform.startswith("win"):
-        args = ("cmd", "/c " + command)
-    else:
-        args = ("/bin/sh", "-c", command)
-
+    args = [azd_path] + command_args
     working_directory = get_safe_working_dir()
 
     try:
+        _LOGGER.debug("Executing subprocess with the following arguments %s", args)
         proc = await asyncio.create_subprocess_exec(
             *args,
             stdout=asyncio.subprocess.PIPE,

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

@@ -3,6 +3,7 @@
 # Licensed under the MIT License.
 # ------------------------------------
 import asyncio
+import logging
 import os
 import shutil
 import sys
@@ -33,6 +34,9 @@
 )
 
 
+_LOGGER = logging.getLogger(__name__)
+
+
 class AzureCliCredential(AsyncContextManager):
     """Authenticates by requesting a token from the Azure CLI.
 
@@ -145,7 +149,7 @@ async def _get_token_base(
             validate_scope(scope)
 
         resource = _scopes_to_resource(*scopes)
-        command = COMMAND_LINE.format(resource)
+        command_args = COMMAND_LINE + ["--resource", resource]
         tenant = resolve_tenant(
             default_tenant=self.tenant_id,
             tenant_id=tenant_id,
@@ -154,10 +158,11 @@ async def _get_token_base(
         )
 
         if tenant:
-            command += " --tenant " + tenant
+            command_args += ["--tenant", tenant]
+
         if self.subscription:
-            command += f' --subscription "{self.subscription}"'
-        output = await _run_command(command, self._process_timeout)
+            command_args += ["--subscription", self.subscription]
+        output = await _run_command(command_args, self._process_timeout)
 
         token = parse_token(output)
         if not token:
@@ -177,19 +182,16 @@ async def close(self) -> None:
         """Calling this method is unnecessary"""
 
 
-async def _run_command(command: str, timeout: int) -> str:
+async def _run_command(command_args: List[str], timeout: int) -> str:
     # Ensure executable exists in PATH first. This avoids a subprocess call that would fail anyway.
-    if shutil.which(EXECUTABLE_NAME) is None:
+    az_path = shutil.which(EXECUTABLE_NAME)
+    if not az_path:
         raise CredentialUnavailableError(message=CLI_NOT_FOUND)
 
-    if sys.platform.startswith("win"):
-        args = ("cmd", "/c " + command)
-    else:
-        args = ("/bin/sh", "-c", command)
-
+    args = [az_path] + command_args
     working_directory = get_safe_working_dir()
-
     try:
+        _LOGGER.debug("Executing subprocess with the following arguments %s", args)
         proc = await asyncio.create_subprocess_exec(
             *args,
             stdout=asyncio.subprocess.PIPE,

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

@@ -211,8 +211,8 @@ def test_multitenant_authentication_class(get_token_method):
     second_token = first_token * 2
 
     def fake_check_output(command_line, **_):
-        match = re.search("--tenant-id (.*)", command_line[-1])
-        tenant = match.groups()[0] if match else default_tenant
+        tenant_id_index = command_line.index("--tenant-id") if "--tenant-id" in command_line else None
+        tenant = command_line[tenant_id_index + 1] if tenant_id_index is not None else default_tenant
         assert tenant in (default_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
         return json.dumps(
             {
@@ -244,8 +244,8 @@ def test_multitenant_authentication(get_token_method):
     second_token = first_token * 2
 
     def fake_check_output(command_line, **_):
-        match = re.search("--tenant-id (.*)", command_line[-1])
-        tenant = match.groups()[0] if match else default_tenant
+        tenant_id_index = command_line.index("--tenant-id") if "--tenant-id" in command_line else None
+        tenant = command_line[tenant_id_index + 1] if tenant_id_index is not None else default_tenant
         assert tenant in (default_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
         return json.dumps(
             {

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

@@ -234,8 +234,8 @@ async def test_multitenant_authentication(get_token_method):
     second_token = first_token * 2
 
     async def fake_exec(*args, **_):
-        match = re.search("--tenant-id (.*)", args[-1])
-        tenant = match[1] if match else default_tenant
+        tenant_id_index = args.index("--tenant-id") if "--tenant-id" in args else None
+        tenant = args[tenant_id_index + 1] if tenant_id_index is not None else default_tenant
         assert tenant in (default_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
         output = json.dumps(
             {
@@ -277,8 +277,9 @@ async def test_multitenant_authentication_not_allowed(get_token_method):
     expected_token = "***"
 
     async def fake_exec(*args, **_):
-        match = re.search("--tenant-id (.*)", args[-1])
-        assert match is None or match[1] == expected_tenant
+        tenant_id_index = args.index("--tenant-id") if "--tenant-id" in args else None
+        tenant = args[tenant_id_index + 1] if tenant_id_index is not None else None
+        assert tenant is None or tenant == expected_tenant
         output = json.dumps(
             {
                 "expiresOn": datetime.now().strftime("%Y-%m-%dT%H:%M:%SZ"),

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

@@ -85,7 +85,9 @@ def test_subscription(get_token_method):
     assert credential.subscription == subscription
 
     def fake_check_output(command_line, **_):
-        assert f'--subscription "{subscription}"' in command_line[-1]
+        assert "--subscription" in command_line
+        subscription_id_index = command_line.index("--subscription")
+        assert command_line[subscription_id_index + 1]
         return json.dumps(
             {
                 "expiresOn": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),
@@ -293,8 +295,8 @@ def test_multitenant_authentication_class(get_token_method):
     second_token = first_token * 2
 
     def fake_check_output(command_line, **_):
-        match = re.search("--tenant (.*)", command_line[-1])
-        tenant = match.groups()[0] if match else default_tenant
+        tenant_index = command_line.index("--tenant") if "--tenant" in command_line else None
+        tenant = command_line[tenant_index + 1] if tenant_index is not None else default_tenant
         assert tenant in (default_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
         return json.dumps(
             {
@@ -326,8 +328,8 @@ def test_multitenant_authentication(get_token_method):
     second_token = first_token * 2
 
     def fake_check_output(command_line, **_):
-        match = re.search("--tenant (.*)", command_line[-1])
-        tenant = match.groups()[0] if match else default_tenant
+        tenant_index = command_line.index("--tenant") if "--tenant" in command_line else None
+        tenant = command_line[tenant_index + 1] if tenant_index is not None else default_tenant
         assert tenant in (default_tenant, second_tenant), 'unexpected tenant "{}"'.format(tenant)
         return json.dumps(
             {
@@ -368,8 +370,9 @@ def test_multitenant_authentication_not_allowed(get_token_method):
     expected_token = "***"
 
     def fake_check_output(command_line, **_):
-        match = re.search("--tenant (.*)", command_line[-1])
-        assert match is None or match[1] == expected_tenant
+        tenant_index = command_line.index("--tenant") if "--tenant" in command_line else None
+        tenant = command_line[tenant_index + 1] if tenant_index is not None else None
+        assert tenant is None or tenant == expected_tenant
         return json.dumps(
             {
                 "expiresOn": datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f"),