feat: Add shlex to correctly parse executable commands with spaces (#1855)

The `subprocess.run` command was using `.split()` which does not handle
quoted paths with spaces correctly. This would cause a
`FileNotFoundError` when the path to the executable contained spaces.

This change replaces `.split()` with `shlex.split()` to correctly parse
the command string.

A test case has been added to verify the fix and prevent regressions.

This was reported in b/237606033

Co-authored-by: Daniel Sanche <d.sanche14@gmail.com>

Author: nbayati

google/auth/pluggable.py

@@ -37,6 +37,7 @@
     from collections import Mapping  # type: ignore
 import json
 import os
+import shlex
 import subprocess
 import sys
 import time
@@ -220,7 +221,7 @@ def retrieve_subject_token(self, request):
         exe_stderr = sys.stdout if self.interactive else subprocess.STDOUT
 
         result = subprocess.run(
-            self._credential_source_executable_command.split(),
+            shlex.split(self._credential_source_executable_command),
             timeout=exe_timeout,
             stdin=exe_stdin,
             stdout=exe_stdout,
@@ -273,7 +274,7 @@ def revoke(self, request):
 
         # Run executable
         result = subprocess.run(
-            self._credential_source_executable_command.split(),
+            shlex.split(self._credential_source_executable_command),
             timeout=self._credential_source_executable_interactive_timeout_millis
             / 1000,
             stdout=subprocess.PIPE,

tests/test_pluggable.py

@@ -1239,6 +1239,36 @@ def test_retrieve_subject_token_python_2(self):
 
             assert excinfo.match(r"Pluggable auth is only supported for python 3.7+")
 
+    @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
+    def test_retrieve_subject_token_with_quoted_command(self):
+        command_with_spaces = '"/path/with spaces/to/executable" "arg with spaces"'
+        credential_source = {
+            "executable": {"command": command_with_spaces, "timeout_millis": 30000}
+        }
+
+        with mock.patch(
+            "subprocess.run",
+            return_value=subprocess.CompletedProcess(
+                args=[],
+                stdout=json.dumps(
+                    self.EXECUTABLE_SUCCESSFUL_OIDC_RESPONSE_ID_TOKEN
+                ).encode("UTF-8"),
+                returncode=0,
+            ),
+        ) as mock_run:
+            credentials = self.make_pluggable(credential_source=credential_source)
+            subject_token = credentials.retrieve_subject_token(None)
+
+            assert subject_token == self.EXECUTABLE_OIDC_TOKEN
+            mock_run.assert_called_once_with(
+                ["/path/with spaces/to/executable", "arg with spaces"],
+                timeout=30.0,
+                stdin=None,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.STDOUT,
+                env=mock.ANY,
+            )
+
     @mock.patch.dict(os.environ, {"GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES": "1"})
     def test_revoke_subject_token_python_2(self):
         with mock.patch("sys.version_info", (2, 7)):