feat: implement pluggable auth interactive mode

google/auth/pluggable.py

@@ -38,6 +38,7 @@
 import json
 import os
 import subprocess
+import sys
 import time
 
 from google.auth import _helpers
@@ -116,6 +117,9 @@ def __init__(
         self._credential_source_executable_timeout_millis = self._credential_source_executable.get(
             "timeout_millis"
         )
+        self._credential_source_executable_interactive_timeout_millis = self._credential_source_executable.get(
+            "interactive_timeout_millis"
+        )
         self._credential_source_executable_output_file = self._credential_source_executable.get(
             "output_file"
         )
@@ -132,6 +136,18 @@ def __init__(
         ):
             raise ValueError("Timeout must be between 5 and 120 seconds.")
 
+        if not self._credential_source_executable_interactive_timeout_millis:
+            self._credential_source_executable_interactive_timeout_millis = (
+                5 * 60 * 1000
+            )
+        elif (
+            self._credential_source_executable_interactive_timeout_millis
+            < 5 * 60 * 1000
+            or self._credential_source_executable_interactive_timeout_millis
+            > 30 * 60 * 1000
+        ):
+            raise ValueError("Interactive timeout must be between 5 and 30 minutes.")
+
     @_helpers.copy_docstring(external_account.Credentials)
     def retrieve_subject_token(self, request):
         env_allow_executables = os.environ.get(
@@ -141,12 +157,13 @@ def retrieve_subject_token(self, request):
             raise ValueError(
                 "Executables need to be explicitly allowed (set GOOGLE_EXTERNAL_ACCOUNT_ALLOW_EXECUTABLES to '1') to run."
             )
+        interactive_mode = os.environ.get("GOOGLE_EXTERNAL_ACCOUNT_INTERACTIVE") == "1"
 
         # Check output file.
         if self._credential_source_executable_output_file is not None:
             try:
                 with open(
-                    self._credential_source_executable_output_file
+                    self._credential_source_executable_output_file, encoding="utf-8"
                 ) as output_file:
                     response = json.load(output_file)
             except Exception:
@@ -155,6 +172,10 @@ def retrieve_subject_token(self, request):
                 try:
                     # If the cached response is expired, _parse_subject_token will raise an error which will be ignored and we will call the executable again.
                     subject_token = self._parse_subject_token(response)
+                    if (
+                        interactive_mode and "expiration_time" not in response
+                    ):  # Always treat missing expiration_time as expired and proceed to executable run.
+                        raise exceptions.RefreshError
                 except ValueError:
                     raise
                 except exceptions.RefreshError:
@@ -171,9 +192,7 @@ def retrieve_subject_token(self, request):
         env = os.environ.copy()
         env["GOOGLE_EXTERNAL_ACCOUNT_AUDIENCE"] = self._audience
         env["GOOGLE_EXTERNAL_ACCOUNT_TOKEN_TYPE"] = self._subject_token_type
-        env[
-            "GOOGLE_EXTERNAL_ACCOUNT_INTERACTIVE"
-        ] = "0"  # Always set to 0 until interactive mode is implemented.
+
         if self._service_account_impersonation_url is not None:
             env[
                 "GOOGLE_EXTERNAL_ACCOUNT_IMPERSONATED_EMAIL"
@@ -183,30 +202,38 @@ def retrieve_subject_token(self, request):
                 "GOOGLE_EXTERNAL_ACCOUNT_OUTPUT_FILE"
             ] = self._credential_source_executable_output_file
 
-        try:
-            result = subprocess.run(
-                self._credential_source_executable_command.split(),
-                timeout=self._credential_source_executable_timeout_millis / 1000,
-                stdout=subprocess.PIPE,
-                stderr=subprocess.STDOUT,
-                env=env,
-            )
-            if result.returncode != 0:
-                raise exceptions.RefreshError(
-                    "Executable exited with non-zero return code {}. Error: {}".format(
-                        result.returncode, result.stdout
-                    )
+        exe_timeout = (
+            self._credential_source_executable_interactive_timeout_millis / 1000
+            if interactive_mode
+            else self._credential_source_executable_timeout_millis / 1000
+        )
+        exe_stdin = sys.stdin if interactive_mode else None
+        exe_stdout = sys.stdout if interactive_mode else subprocess.PIPE
+        exe_stderr = sys.stdout if interactive_mode else subprocess.STDOUT
+
+        result = subprocess.run(
+            self._credential_source_executable_command.split(),
+            timeout=exe_timeout / 1000,
+            stdin=exe_stdin,
+            stdout=exe_stdout,
+            stderr=exe_stderr,
+            env=env,
+        )
+        if result.returncode != 0:
+            raise exceptions.RefreshError(
+                "Executable exited with non-zero return code {}. Error: {}".format(
+                    result.returncode, result.stdout
                 )
-        except Exception:
-            raise
-        else:
-            try:
-                data = result.stdout.decode("utf-8")
-                response = json.loads(data)
-                subject_token = self._parse_subject_token(response)
-            except Exception:
-                raise
+            )
 
+        response = (
+            json.load(
+                open(self._credential_source_executable_output_file, encoding="utf-8")
+            )
+            if interactive_mode
+            else json.loads(result.stdout.decode("utf-8"))
+        )
+        subject_token = self._parse_subject_token(response)
         return subject_token
 
     @classmethod
@@ -264,10 +291,11 @@ def _parse_subject_token(self, response):
             )
         if (
             "expiration_time" not in response
+            and not os.environ.get("GOOGLE_EXTERNAL_ACCOUNT_INTERACTIVE") == "1"
             and self._credential_source_executable_output_file
         ):
             raise ValueError(
-                "The executable response must contain an expiration_time for successful responses when an output_file has been specified in the configuration."
+                "The executable response must contain an expiration_time for successful responses when an output_file has been specified in the configuration in non-interactive mode."
             )
         if "expiration_time" in response and response["expiration_time"] < time.time():
             raise exceptions.RefreshError(