feat: support external mode

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'
 ]

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
+        )

alibabacloud_credentials/provider/external.py

@@ -0,0 +1,165 @@
+import asyncio
+import calendar
+import json
+import logging
+import os
+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]:
+        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(
+                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')
+        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]:
+        if not self._process_command.strip():
+            raise CredentialException('process_command is empty')
+
+        try:
+            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():
+                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'