Update subprocess arguments to use expanded user path

awsume/awsumepy/default_plugins.py

@@ -2,6 +2,7 @@
 import json
 import os
 import subprocess
+from pathlib import Path
 
 import colorama
 import dateutil
@@ -519,7 +520,8 @@ def get_credentials_from_credential_process(config: dict, arguments: argparse.Na
     return_session = {}
     credential_process_env = os.environ.copy()
     credential_process_env['AWS_PROFILE'] = target_profile_name
-    result = subprocess.run(target_profile.get('credential_process').split(), stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=credential_process_env)
+    credential_process_target_and_arguments = get_credentials_process_target_and_arguments(target_profile)
+    result = subprocess.run(credential_process_target_and_arguments, stdout=subprocess.PIPE, stderr=subprocess.PIPE, env=credential_process_env)
     logger.info('credential_process returncode: {}'.format(result.returncode))
     logger.debug('credential_process stdout: {}'.format(result.stdout.decode('utf-8')))
     logger.debug('credential_process stderr: {}'.format(result.stderr.decode('utf-8')))
@@ -547,6 +549,20 @@ def get_credentials_from_credential_process(config: dict, arguments: argparse.Na
     logger.debug("credential_process session: {}".format(return_session))
     return return_session
 
+def get_credentials_process_target_and_arguments(target_profile: dict):
+    credential_process = target_profile.get("credential_process", None)
+    if credential_process is None:
+        raise exceptions.ValidationException(f"credential_process not found in profile: {json.dumps(target_profile)}")
+    parts = credential_process.split()
+    target = Path(parts[0]).expanduser() if len(parts) > 0 else None
+    if target is None:
+        logger.debug(f"Unable to find credentials target from provided process string: {credential_process}")
+        raise exceptions.ValidationException('credential_process target not found: {}'.format(target))
+    if not target.is_file():
+        logger.debug(f"{credential_process} is not a file. Derived target is {str(target)}")
+        raise exceptions.ValidationException('credential_process target is not a file: {}'.format(target))
+    return [str(target), *parts[1:]]
+
 
 def get_session_token_credentials(config: dict, arguments: argparse.Namespace, profiles: dict, target_profile: dict, target_profile_name: str):
     logger.info('Getting session token credentials')

test/unit/awsume/awsumepy/test_default_plugins.py

@@ -1,9 +1,12 @@
 import argparse
+from pathlib import Path
+
 import pytest
 from io import StringIO
 from unittest.mock import MagicMock, patch
 from awsume.awsumepy import default_plugins
 from awsume.awsumepy.lib import exceptions, autoawsume
+from awsume.awsumepy.lib.constants import AWSUME_DIR
 
 
 def generate_namespace_with_defaults(
@@ -1006,3 +1009,46 @@ def test_post_add_arguments_session_tags(aws_lib: MagicMock):
         ],
         region=arguments.region,
     )
+
+@patch.object(Path, 'is_file')
+def test_get_credentials_process_target_and_arguments_with_file(is_file: MagicMock):
+    is_file.return_value(True)
+    process_file = Path(f"{AWSUME_DIR}/test.sh").expanduser()
+    expected = [str(process_file), "arg1", "arg2"]
+    target_profile = {
+        "credential_process": f"{str(process_file)} arg1 arg2"
+    }
+    actual = default_plugins.get_credentials_process_target_and_arguments(target_profile)
+    assert actual == expected
+
+def test_get_credentials_process_target_and_arguments_without_file():
+    process_file = Path(f"{AWSUME_DIR}/nonexistent.sh").expanduser()
+    target_profile = {
+        "credential_process": f"{str(process_file)} arg1 arg2"
+    }
+    with pytest.raises(exceptions.ValidationException):
+        default_plugins.get_credentials_process_target_and_arguments(target_profile)
+
+def test_get_credentials_process_target_and_arguments_invalid_arguments():
+    target_profile = {
+        "credential_process": ""
+    }
+    with pytest.raises(exceptions.ValidationException):
+        default_plugins.get_credentials_process_target_and_arguments(target_profile)
+
+def test_get_credentials_process_target_and_arguments_invalid_input():
+    target_profile = {}
+    with pytest.raises(exceptions.ValidationException):
+        default_plugins.get_credentials_process_target_and_arguments(target_profile)
+
+def test_get_credentials_process_target_and_arguments_expands_user():
+    process_file = Path(f"~/test.sh")
+    expanded_process_file = process_file.expanduser()
+    expanded_process_file.open('w').close()
+    target_profile = {
+        "credential_process": f"{str(process_file)} arg1 arg2"
+    }
+    print(target_profile)
+    expected = [str(expanded_process_file), "arg1", "arg2"]
+    actual = default_plugins.get_credentials_process_target_and_arguments(target_profile)
+    assert actual == expected