From d858520e836443c04974c68f8f0184a18bb4e96e Mon Sep 17 00:00:00 2001 From: yndu13 Date: Tue, 9 Jun 2026 18:08:07 +0800 Subject: [PATCH 1/2] feat: support external mode --- alibabacloud_credentials/provider/__init__.py | 4 +- .../provider/cli_profile.py | 95 +++++++++++ alibabacloud_credentials/provider/external.py | 157 ++++++++++++++++++ tests/provider/test_cli_profile.py | 112 ++++++++++++- tests/provider/test_external.py | 143 ++++++++++++++++ 5 files changed, 509 insertions(+), 2 deletions(-) create mode 100644 alibabacloud_credentials/provider/external.py create mode 100644 tests/provider/test_external.py diff --git a/alibabacloud_credentials/provider/__init__.py b/alibabacloud_credentials/provider/__init__.py index 60eaf60..a9d9d84 100644 --- a/alibabacloud_credentials/provider/__init__.py +++ b/alibabacloud_credentials/provider/__init__.py @@ -11,6 +11,7 @@ from .default import DefaultCredentialsProvider from .cloud_sso import CloudSSOCredentialsProvider from .oauth import OAuthCredentialsProvider +from .external import ExternalCredentialsProvider __all__ = [ 'StaticAKCredentialsProvider', @@ -25,5 +26,6 @@ 'ProfileCredentialsProvider', 'DefaultCredentialsProvider', 'CloudSSOCredentialsProvider', - 'OAuthCredentialsProvider' + 'OAuthCredentialsProvider', + 'ExternalCredentialsProvider' ] diff --git a/alibabacloud_credentials/provider/cli_profile.py b/alibabacloud_credentials/provider/cli_profile.py index e04cef4..bfa5fb5 100644 --- a/alibabacloud_credentials/provider/cli_profile.py +++ b/alibabacloud_credentials/provider/cli_profile.py @@ -30,6 +30,11 @@ from .static_sts import StaticSTSCredentialsProvider from .cloud_sso import CloudSSOCredentialsProvider from .oauth import OAuthCredentialsProvider, OAuthTokenUpdateCallback, OAuthTokenUpdateCallbackAsync +from .external import ( + ExternalCredentialsProvider, + ExternalCredentialUpdateCallback, + ExternalCredentialUpdateCallbackAsync, +) from .refreshable import Credentials from alibabacloud_credentials_api import ICredentialsProvider from alibabacloud_credentials.utils import auth_constant as ac @@ -228,6 +233,12 @@ def _get_credentials_provider(self, config: Dict, profile_name: str) -> ICredent token_update_callback=self._get_oauth_token_update_callback(), token_update_callback_async=self._get_oauth_token_update_callback_async(), ) + elif mode == "External": + return ExternalCredentialsProvider( + process_command=profile.get('process_command'), + credential_update_callback=self._get_external_credential_update_callback(), + credential_update_callback_async=self._get_external_credential_update_callback_async(), + ) else: raise CredentialException(f"unsupported profile mode '{mode}' form cli credentials file.") @@ -421,6 +432,48 @@ def _get_oauth_token_update_callback(self) -> OAuthTokenUpdateCallback: refresh_token, access_token, access_key, secret, security_token, access_token_expire, sts_expire ) + def _update_external_credentials(self, access_key: str, secret: str, + security_token: str, expiration: int) -> None: + """更新 External 凭证并写回配置文件""" + + def _find_source_external_profile(config: dict, profile_name: str) -> dict: + profiles = config.get('profiles', []) + profile = next((p for p in profiles if p.get('name') == profile_name), None) + if not profile: + raise CredentialException(f"unable to get profile with name '{profile_name}' from cli credentials file.") + + if profile.get('mode') == 'External': + return profile + + source_profile = profile.get('source_profile') + if source_profile: + return _find_source_external_profile(config, source_profile) + + raise CredentialException(f"unable to get External profile with name '{profile_name}' from cli credentials file.") + + with self._file_lock: + try: + config = _load_config(self._profile_file) + profile_name = self._profile_name or config.get('current') + if not profile_name: + raise CredentialException(f"unable to get profile to updated.") + + source_profile = _find_source_external_profile(config, profile_name) + source_profile['access_key_id'] = access_key + source_profile['access_key_secret'] = secret + source_profile['sts_token'] = security_token + source_profile['sts_expiration'] = expiration + + self._write_configuration_to_file_with_lock(self._profile_file, config) + except Exception as e: + raise CredentialException(f"failed to update External credentials in config file: {e}") + + def _get_external_credential_update_callback(self) -> ExternalCredentialUpdateCallback: + """获取 External 凭证更新回调函数""" + return lambda access_key, secret, security_token, expiration: self._update_external_credentials( + access_key, secret, security_token, expiration + ) + async def _write_configuration_to_file_async(self, config_path: str, config: Dict) -> None: """异步将配置写入文件，使用原子写入确保数据完整性""" # 获取原文件权限（如果存在） @@ -604,3 +657,45 @@ def _get_oauth_token_update_callback_async(self) -> OAuthTokenUpdateCallbackAsyn return lambda refresh_token, access_token, access_key, secret, security_token, access_token_expire, sts_expire: self._update_oauth_tokens_async( refresh_token, access_token, access_key, secret, security_token, access_token_expire, sts_expire ) + + async def _update_external_credentials_async(self, access_key: str, secret: str, + security_token: str, expiration: int) -> None: + """异步更新 External 凭证并写回配置文件""" + + def _find_source_external_profile(config: dict, profile_name: str) -> dict: + profiles = config.get('profiles', []) + profile = next((p for p in profiles if p.get('name') == profile_name), None) + if not profile: + raise CredentialException(f"unable to get profile with name '{profile_name}' from cli credentials file.") + + if profile.get('mode') == 'External': + return profile + + source_profile = profile.get('source_profile') + if source_profile: + return _find_source_external_profile(config, source_profile) + + raise CredentialException(f"unable to get External profile with name '{profile_name}' from cli credentials file.") + + with self._file_lock: + try: + config = await _load_config_async(self._profile_file) + profile_name = self._profile_name or config.get('current') + if not profile_name: + raise CredentialException(f"unable to get profile to updated.") + + source_profile = _find_source_external_profile(config, profile_name) + source_profile['access_key_id'] = access_key + source_profile['access_key_secret'] = secret + source_profile['sts_token'] = security_token + source_profile['sts_expiration'] = expiration + + await self._write_configuration_to_file_with_lock_async(self._profile_file, config) + except Exception as e: + raise CredentialException(f"failed to update External credentials in config file: {e}") + + def _get_external_credential_update_callback_async(self) -> ExternalCredentialUpdateCallbackAsync: + """获取异步 External 凭证更新回调函数""" + return lambda access_key, secret, security_token, expiration: self._update_external_credentials_async( + access_key, secret, security_token, expiration + ) diff --git a/alibabacloud_credentials/provider/external.py b/alibabacloud_credentials/provider/external.py new file mode 100644 index 0000000..cee927d --- /dev/null +++ b/alibabacloud_credentials/provider/external.py @@ -0,0 +1,157 @@ +import asyncio +import calendar +import json +import logging +import shlex +import subprocess +import time +from typing import Callable, Optional + +from alibabacloud_credentials.provider.refreshable import Credentials, RefreshResult, RefreshCachedSupplier +from alibabacloud_credentials_api import ICredentialsProvider +from alibabacloud_credentials.exceptions import CredentialException + +log = logging.getLogger('credentials') + +ExternalCredentialUpdateCallback = Callable[[str, str, str, int], None] +ExternalCredentialUpdateCallbackAsync = Callable[[str, str, str, int], None] + + +def _parse_expiration(expiration: str) -> int: + if not expiration: + return 0 + time_array = time.strptime(expiration, '%Y-%m-%dT%H:%M:%SZ') + return calendar.timegm(time_array) + + +def _get_stale_time(expiration: int) -> int: + if expiration <= 0: + return int(time.mktime(time.localtime())) + return expiration - 180 + + +class ExternalCredentialsProvider(ICredentialsProvider): + DEFAULT_TIMEOUT = 60 + + def __init__(self, *, + process_command: str = None, + timeout: int = None, + credential_update_callback: Optional[ExternalCredentialUpdateCallback] = None, + credential_update_callback_async: Optional[ExternalCredentialUpdateCallbackAsync] = None): + if not process_command: + raise ValueError('process_command is empty') + + self._process_command = process_command + self._timeout = timeout if timeout and timeout > 0 else ExternalCredentialsProvider.DEFAULT_TIMEOUT + self._credential_update_callback = credential_update_callback + self._credential_update_callback_async = credential_update_callback_async + self._credentials_cache = RefreshCachedSupplier( + refresh_callable=self._refresh_credentials, + refresh_callable_async=self._refresh_credentials_async, + ) + + def get_credentials(self) -> Credentials: + return self._credentials_cache._sync_call() + + async def get_credentials_async(self) -> Credentials: + return await self._credentials_cache._async_call() + + def _refresh_credentials(self) -> RefreshResult[Credentials]: + args = shlex.split(self._process_command) + if not args: + raise CredentialException('process_command is empty') + + try: + completed = subprocess.run( + args, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + timeout=self._timeout, + check=False, + text=True, + ) + except subprocess.TimeoutExpired: + raise CredentialException(f'command process timed out after {self._timeout * 1000} milliseconds') + except Exception as e: + raise CredentialException(f'failed to execute external command: {e}') + + if completed.returncode != 0: + raise CredentialException( + f'failed to execute external command: exit status {completed.returncode}\nstderr: {completed.stderr}') + + return self._parse_and_build_credentials(completed.stdout, async_callback=False) + + async def _refresh_credentials_async(self) -> RefreshResult[Credentials]: + args = shlex.split(self._process_command) + if not args: + raise CredentialException('process_command is empty') + + try: + process = await asyncio.create_subprocess_exec( + *args, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=self._timeout) + except asyncio.TimeoutError: + if 'process' in locals(): + process.kill() + await process.wait() + raise CredentialException(f'command process timed out after {self._timeout * 1000} milliseconds') + except Exception as e: + raise CredentialException(f'failed to execute external command: {e}') + + if process.returncode != 0: + raise CredentialException( + f'failed to execute external command: exit status {process.returncode}\nstderr: {stderr.decode("utf-8")}') + + return await self._parse_and_build_credentials_async(stdout.decode('utf-8')) + + def _parse_and_build_credentials(self, output: str, async_callback: bool) -> RefreshResult[Credentials]: + try: + data = json.loads(output) + except Exception as e: + raise CredentialException(f'failed to parse external command output: {e}') + + access_key_id = data.get('access_key_id') + access_key_secret = data.get('access_key_secret') + security_token = data.get('sts_token') + if not access_key_id or not access_key_secret: + raise CredentialException('invalid credential response: access_key_id or access_key_secret is empty') + if data.get('mode') == 'StsToken' and not security_token: + raise CredentialException('invalid StsToken credential response: sts_token is empty') + + expiration = _parse_expiration(data.get('expiration')) + credentials = Credentials( + access_key_id=access_key_id, + access_key_secret=access_key_secret, + security_token=security_token, + expiration=expiration, + provider_name=self.get_provider_name(), + ) + + if not async_callback and self._credential_update_callback: + try: + self._credential_update_callback(access_key_id, access_key_secret, security_token, expiration) + except Exception as e: + log.warning(f'failed to update external credentials in config file: {e}') + + return RefreshResult(value=credentials, stale_time=_get_stale_time(expiration)) + + async def _parse_and_build_credentials_async(self, output: str) -> RefreshResult[Credentials]: + result = self._parse_and_build_credentials(output, async_callback=True) + credentials = result.value() + if self._credential_update_callback_async: + try: + await self._credential_update_callback_async( + credentials.get_access_key_id(), + credentials.get_access_key_secret(), + credentials.get_security_token(), + credentials.get_expiration() or 0, + ) + except Exception as e: + log.warning(f'failed to update external credentials in config file: {e}') + return result + + def get_provider_name(self) -> str: + return 'external' diff --git a/tests/provider/test_cli_profile.py b/tests/provider/test_cli_profile.py index f765b8d..34db8c4 100644 --- a/tests/provider/test_cli_profile.py +++ b/tests/provider/test_cli_profile.py @@ -16,7 +16,8 @@ EcsRamRoleCredentialsProvider, OIDCRoleArnCredentialsProvider, CloudSSOCredentialsProvider, - OAuthCredentialsProvider + OAuthCredentialsProvider, + ExternalCredentialsProvider ) from alibabacloud_credentials.utils import auth_constant as ac @@ -102,6 +103,11 @@ def setUp(self): "oauth_refresh_token": "test_refresh_token", "oauth_access_token": "test_oauth_access_token", "oauth_access_token_expire": int(time.mktime(time.localtime())) + 1000 + }, + { + "name": "external_profile", + "mode": "External", + "process_command": "/bin/echo '{\"mode\":\"AK\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'" } ] } @@ -306,6 +312,25 @@ def test_get_credentials_valid_oauth(self): self.assertEqual(credentials_provider._access_token, 'test_oauth_access_token') self.assertTrue(credentials_provider._access_token_expire > int(time.mktime(time.localtime()))) + def test_get_credentials_valid_external(self): + """ + Test case 9: Valid input, successfully retrieves credentials for External mode + """ + with patch('alibabacloud_credentials.provider.cli_profile.au.environment_cli_profile_disabled', False): + with patch('os.path.exists', return_value=True): + with patch('os.path.isfile', return_value=True): + with patch('alibabacloud_credentials.provider.cli_profile._load_config', return_value=self.config): + provider = CLIProfileCredentialsProvider(profile_name="external_profile") + + credentials_provider = provider._get_credentials_provider(config=self.config, + profile_name="external_profile") + + self.assertIsInstance(credentials_provider, ExternalCredentialsProvider) + self.assertEqual( + credentials_provider._process_command, + "/bin/echo '{\"mode\":\"AK\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'" + ) + def test_get_credentials_cli_profile_disabled(self): """ Test case 9: CLI profile disabled raises CredentialException @@ -547,6 +572,91 @@ def test_oauth_token_update_callback(self): import shutil shutil.rmtree(temp_dir, ignore_errors=True) + def test_update_external_credentials(self): + """测试 External 凭证更新回调写回当前 External profile""" + import tempfile + import shutil + + temp_dir = tempfile.mkdtemp() + config_path = os.path.join(temp_dir, 'config.json') + test_config = { + "current": "external_source", + "profiles": [ + { + "name": "external_source", + "mode": "External", + "process_command": "echo test" + } + ] + } + + with open(config_path, 'w') as f: + json.dump(test_config, f, indent=4) + + try: + provider = CLIProfileCredentialsProvider( + profile_name="external_source", + profile_file=config_path, + allow_config_force_rewrite=True + ) + + provider._update_external_credentials("new_ak", "new_secret", "new_token", 4102444800) + + with open(config_path, 'r') as f: + updated_config = json.load(f) + + external_profile = updated_config['profiles'][0] + self.assertEqual(external_profile['access_key_id'], "new_ak") + self.assertEqual(external_profile['access_key_secret'], "new_secret") + self.assertEqual(external_profile['sts_token'], "new_token") + self.assertEqual(external_profile['sts_expiration'], 4102444800) + finally: + shutil.rmtree(temp_dir, ignore_errors=True) + + def test_update_external_credentials_async(self): + """测试异步 External 凭证更新回调写回当前 External profile""" + async def run_test(): + import tempfile + import shutil + + temp_dir = tempfile.mkdtemp() + config_path = os.path.join(temp_dir, 'config.json') + test_config = { + "current": "external_source", + "profiles": [ + { + "name": "external_source", + "mode": "External", + "process_command": "echo test" + } + ] + } + + with open(config_path, 'w') as f: + json.dump(test_config, f, indent=4) + + try: + provider = CLIProfileCredentialsProvider( + profile_name="external_source", + profile_file=config_path, + allow_config_force_rewrite=True + ) + + await provider._update_external_credentials_async("async_ak", "async_secret", "async_token", 4102444801) + + with open(config_path, 'r') as f: + updated_config = json.load(f) + + external_profile = updated_config['profiles'][0] + self.assertEqual(external_profile['access_key_id'], "async_ak") + self.assertEqual(external_profile['access_key_secret'], "async_secret") + self.assertEqual(external_profile['sts_token'], "async_token") + self.assertEqual(external_profile['sts_expiration'], 4102444801) + finally: + shutil.rmtree(temp_dir, ignore_errors=True) + + asyncio.run(run_test()) + def test_oauth_callback_integration(self): """测试 OAuth 回调集成""" import tempfile diff --git a/tests/provider/test_external.py b/tests/provider/test_external.py new file mode 100644 index 0000000..4df8cfa --- /dev/null +++ b/tests/provider/test_external.py @@ -0,0 +1,143 @@ +import asyncio +import json +import os +import tempfile +import time +import unittest + +from alibabacloud_credentials.exceptions import CredentialException +from alibabacloud_credentials.provider.external import ExternalCredentialsProvider, _get_stale_time + + +class TestExternalCredentialsProvider(unittest.TestCase): + + def _create_script(self, content): + temp_dir = tempfile.mkdtemp() + script_path = os.path.join(temp_dir, 'external_credential.sh') + with open(script_path, 'w') as f: + f.write(content) + os.chmod(script_path, 0o755) + return script_path + + def test_init_validation(self): + with self.assertRaises(ValueError) as context: + ExternalCredentialsProvider() + self.assertIn('process_command is empty', str(context.exception)) + + def test_get_credentials_ak_success(self): + script_path = self._create_script( + "#!/bin/sh\n" + "echo '{\"mode\":\"AK\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'\n" + ) + provider = ExternalCredentialsProvider(process_command=script_path) + + credentials = provider.get_credentials() + + self.assertEqual(credentials.get_access_key_id(), 'akid') + self.assertEqual(credentials.get_access_key_secret(), 'secret') + self.assertIsNone(credentials.get_security_token()) + self.assertEqual(credentials.get_provider_name(), 'external') + + def test_get_credentials_sts_success_with_callback(self): + expiration = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(time.time() + 3600)) + script_path = self._create_script( + "#!/bin/sh\n" + "echo '" + json.dumps({ + 'mode': 'StsToken', + 'access_key_id': 'akid', + 'access_key_secret': 'secret', + 'sts_token': 'token', + 'expiration': expiration, + }) + "'\n" + ) + callback_args = [] + provider = ExternalCredentialsProvider( + process_command=script_path, + credential_update_callback=lambda *args: callback_args.append(args), + ) + + credentials = provider.get_credentials() + + self.assertEqual(credentials.get_security_token(), 'token') + self.assertEqual(len(callback_args), 1) + self.assertEqual(callback_args[0][0], 'akid') + self.assertEqual(callback_args[0][1], 'secret') + self.assertEqual(callback_args[0][2], 'token') + self.assertGreater(callback_args[0][3], 0) + + def test_get_credentials_async_success_with_callback(self): + async def run_test(): + expiration = time.strftime('%Y-%m-%dT%H:%M:%SZ', time.gmtime(time.time() + 3600)) + script_path = self._create_script( + "#!/bin/sh\n" + "echo '" + json.dumps({ + 'mode': 'StsToken', + 'access_key_id': 'akid', + 'access_key_secret': 'secret', + 'sts_token': 'token', + 'expiration': expiration, + }) + "'\n" + ) + callback_args = [] + + async def callback(*args): + callback_args.append(args) + + provider = ExternalCredentialsProvider( + process_command=script_path, + credential_update_callback_async=callback, + ) + + credentials = await provider.get_credentials_async() + self.assertEqual(credentials.get_access_key_id(), 'akid') + self.assertEqual(credentials.get_security_token(), 'token') + self.assertEqual(len(callback_args), 1) + + asyncio.run(run_test()) + + def test_invalid_json(self): + script_path = self._create_script("#!/bin/sh\necho 'invalid json'\n") + provider = ExternalCredentialsProvider(process_command=script_path) + + with self.assertRaises(CredentialException) as context: + provider.get_credentials() + self.assertIn('failed to parse external command output', str(context.exception)) + + def test_missing_access_key(self): + script_path = self._create_script( + "#!/bin/sh\n" + "echo '{\"mode\":\"AK\",\"access_key_id\":\"\",\"access_key_secret\":\"secret\"}'\n" + ) + provider = ExternalCredentialsProvider(process_command=script_path) + + with self.assertRaises(CredentialException) as context: + provider.get_credentials() + self.assertIn('access_key_id or access_key_secret is empty', str(context.exception)) + + def test_missing_sts_token(self): + script_path = self._create_script( + "#!/bin/sh\n" + "echo '{\"mode\":\"StsToken\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'\n" + ) + provider = ExternalCredentialsProvider(process_command=script_path) + + with self.assertRaises(CredentialException) as context: + provider.get_credentials() + self.assertIn('sts_token is empty', str(context.exception)) + + def test_command_failure(self): + script_path = self._create_script("#!/bin/sh\necho failed >&2\nexit 1\n") + provider = ExternalCredentialsProvider(process_command=script_path) + + with self.assertRaises(CredentialException) as context: + provider.get_credentials() + self.assertIn('failed to execute external command', str(context.exception)) + + def test_get_stale_time(self): + now = int(time.mktime(time.localtime())) + self.assertLessEqual(_get_stale_time(0), now) + self.assertEqual(_get_stale_time(1000), 820) + + +if __name__ == '__main__': + unittest.main() From 6e37f3593682d693779516d0cfe51dc6f142384b Mon Sep 17 00:00:00 2001 From: yndu13 Date: Tue, 9 Jun 2026 19:31:18 +0800 Subject: [PATCH 2/2] fix --- alibabacloud_credentials/provider/external.py | 28 ++++++---- tests/provider/test_external.py | 54 ++++++++++++++----- 2 files changed, 60 insertions(+), 22 deletions(-) diff --git a/alibabacloud_credentials/provider/external.py b/alibabacloud_credentials/provider/external.py index cee927d..121a67a 100644 --- a/alibabacloud_credentials/provider/external.py +++ b/alibabacloud_credentials/provider/external.py @@ -2,6 +2,7 @@ import calendar import json import logging +import os import shlex import subprocess import time @@ -57,18 +58,19 @@ async def get_credentials_async(self) -> Credentials: return await self._credentials_cache._async_call() def _refresh_credentials(self) -> RefreshResult[Credentials]: - args = shlex.split(self._process_command) - if not args: + if not self._process_command.strip(): raise CredentialException('process_command is empty') try: + command = self._process_command if os.name == 'nt' else shlex.split(self._process_command) completed = subprocess.run( - args, + command, stdout=subprocess.PIPE, stderr=subprocess.PIPE, timeout=self._timeout, check=False, text=True, + shell=os.name == 'nt', ) except subprocess.TimeoutExpired: raise CredentialException(f'command process timed out after {self._timeout * 1000} milliseconds') @@ -82,16 +84,22 @@ def _refresh_credentials(self) -> RefreshResult[Credentials]: return self._parse_and_build_credentials(completed.stdout, async_callback=False) async def _refresh_credentials_async(self) -> RefreshResult[Credentials]: - args = shlex.split(self._process_command) - if not args: + if not self._process_command.strip(): raise CredentialException('process_command is empty') try: - process = await asyncio.create_subprocess_exec( - *args, - stdout=asyncio.subprocess.PIPE, - stderr=asyncio.subprocess.PIPE, - ) + if os.name == 'nt': + process = await asyncio.create_subprocess_shell( + self._process_command, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) + else: + process = await asyncio.create_subprocess_exec( + *shlex.split(self._process_command), + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + ) stdout, stderr = await asyncio.wait_for(process.communicate(), timeout=self._timeout) except asyncio.TimeoutError: if 'process' in locals(): diff --git a/tests/provider/test_external.py b/tests/provider/test_external.py index 4df8cfa..d49195c 100644 --- a/tests/provider/test_external.py +++ b/tests/provider/test_external.py @@ -11,13 +11,15 @@ class TestExternalCredentialsProvider(unittest.TestCase): - def _create_script(self, content): + def _create_script(self, content, windows_content=None): temp_dir = tempfile.mkdtemp() - script_path = os.path.join(temp_dir, 'external_credential.sh') + script_path = os.path.join(temp_dir, 'external_credential.bat' if os.name == 'nt' else 'external_credential.sh') with open(script_path, 'w') as f: - f.write(content) - os.chmod(script_path, 0o755) - return script_path + f.write(windows_content if os.name == 'nt' and windows_content is not None else content) + if os.name != 'nt': + os.chmod(script_path, 0o755) + return script_path + return '"{}"'.format(script_path) def test_init_validation(self): with self.assertRaises(ValueError) as context: @@ -27,7 +29,9 @@ def test_init_validation(self): def test_get_credentials_ak_success(self): script_path = self._create_script( "#!/bin/sh\n" - "echo '{\"mode\":\"AK\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'\n" + "echo '{\"mode\":\"AK\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'\n", + "@echo off\n" + "echo {\"mode\":\"AK\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}\n" ) provider = ExternalCredentialsProvider(process_command=script_path) @@ -48,7 +52,15 @@ def test_get_credentials_sts_success_with_callback(self): 'access_key_secret': 'secret', 'sts_token': 'token', 'expiration': expiration, - }) + "'\n" + }) + "'\n", + "@echo off\n" + "echo " + json.dumps({ + 'mode': 'StsToken', + 'access_key_id': 'akid', + 'access_key_secret': 'secret', + 'sts_token': 'token', + 'expiration': expiration, + }) + "\n" ) callback_args = [] provider = ExternalCredentialsProvider( @@ -76,7 +88,15 @@ async def run_test(): 'access_key_secret': 'secret', 'sts_token': 'token', 'expiration': expiration, - }) + "'\n" + }) + "'\n", + "@echo off\n" + "echo " + json.dumps({ + 'mode': 'StsToken', + 'access_key_id': 'akid', + 'access_key_secret': 'secret', + 'sts_token': 'token', + 'expiration': expiration, + }) + "\n" ) callback_args = [] @@ -96,7 +116,10 @@ async def callback(*args): asyncio.run(run_test()) def test_invalid_json(self): - script_path = self._create_script("#!/bin/sh\necho 'invalid json'\n") + script_path = self._create_script( + "#!/bin/sh\necho 'invalid json'\n", + "@echo off\necho invalid json\n" + ) provider = ExternalCredentialsProvider(process_command=script_path) with self.assertRaises(CredentialException) as context: @@ -106,7 +129,9 @@ def test_invalid_json(self): def test_missing_access_key(self): script_path = self._create_script( "#!/bin/sh\n" - "echo '{\"mode\":\"AK\",\"access_key_id\":\"\",\"access_key_secret\":\"secret\"}'\n" + "echo '{\"mode\":\"AK\",\"access_key_id\":\"\",\"access_key_secret\":\"secret\"}'\n", + "@echo off\n" + "echo {\"mode\":\"AK\",\"access_key_id\":\"\",\"access_key_secret\":\"secret\"}\n" ) provider = ExternalCredentialsProvider(process_command=script_path) @@ -117,7 +142,9 @@ def test_missing_access_key(self): def test_missing_sts_token(self): script_path = self._create_script( "#!/bin/sh\n" - "echo '{\"mode\":\"StsToken\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'\n" + "echo '{\"mode\":\"StsToken\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}'\n", + "@echo off\n" + "echo {\"mode\":\"StsToken\",\"access_key_id\":\"akid\",\"access_key_secret\":\"secret\"}\n" ) provider = ExternalCredentialsProvider(process_command=script_path) @@ -126,7 +153,10 @@ def test_missing_sts_token(self): self.assertIn('sts_token is empty', str(context.exception)) def test_command_failure(self): - script_path = self._create_script("#!/bin/sh\necho failed >&2\nexit 1\n") + script_path = self._create_script( + "#!/bin/sh\necho failed >&2\nexit 1\n", + "@echo off\necho failed 1>&2\nexit /b 1\n" + ) provider = ExternalCredentialsProvider(process_command=script_path) with self.assertRaises(CredentialException) as context: