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